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

@@ -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(),

apps/archive-bot/src/main.ts

@@ -29,7 +29,7 @@ async function main(): Promise<void> {
   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();

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<string>();
   outbox = new Map<string, OutboxRow & { claimKey: string }>();
   archive: (ArchiveRecord & { available: boolean })[] = [];
-  async loadCursor(): Promise<string | null> {
+  async loadCursor(_rooms: readonly RoomCursorTarget[]): Promise<string | null> {
     return this.cursor;
   }
-  async saveCursor(_ws: readonly string[], cursor: string): Promise<void> {
+  async saveCursor(rooms: readonly RoomCursorTarget[], cursor: string): Promise<void> {
+    for (const room of rooms) this.roomsSaved.add(room.roomId);
     this.cursor = cursor;
   }
   async claim(_ws: string, _room: string, _event: string, key: string): Promise<OutboxRow> {
@@ -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 });

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<string, string>();
   private readonly byRoom: Map<string, RoomMapping>;
+  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<void> {
-    this.cursor = await this.store.loadCursor(this.workspaceIds);
+    this.cursor = await this.store.loadCursor(this.cursorRooms);
   }
 
   async health(): Promise<HealthSnapshot> {
@@ -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 };
   }

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<string | null>;
+  loadCursor(rooms: readonly RoomCursorTarget[]): Promise<string | null>;
-  saveCursor(workspaceIds: readonly string[], cursor: string): Promise<void>;
+  saveCursor(rooms: readonly RoomCursorTarget[], cursor: string): Promise<void>;
   claim(workspaceId: string, roomId: string, eventId: string, claimKey: string): Promise<OutboxRow>;
   /** 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(
+  constructor(private readonly pool: Pool) {}
-    private readonly pool: Pool,
-    private readonly cursorId: string,
-  ) {}
 
   /** Single service connection; tenant context per transaction, like apps/api withTenant. */
-  static create(databaseUrl: string, cursorId: string): PgArchiveStore {
+  static create(databaseUrl: string): PgArchiveStore {
-    return new PgArchiveStore(new Pool({ connectionString: databaseUrl, max: 1 }), cursorId);
+    return new PgArchiveStore(new Pool({ connectionString: databaseUrl, max: 1 }));
   }
 
   async close(): Promise<void> {
@@ -72,17 +79,17 @@ export class PgArchiveStore implements ArchiveStore {
   }
 
   /**
-   * The sync stream is global but rows are tenant-scoped, so each mapped
+   * The sync stream is global but cursor rows are per room, so each mapped room
-   * workspace holds a copy. A partial write leaves copies disagreeing: resume
+   * holds a copy. A partial write leaves copies disagreeing: resume from the
-   * from the least recently written (oldest) one — replay is idempotent.
+   * least recently written (oldest) one — replay is idempotent.
    */
-  async loadCursor(workspaceIds: readonly string[]): Promise<string | null> {
+  async loadCursor(rooms: readonly RoomCursorTarget[]): Promise<string | null> {
     let oldest: { cursor: string; at: Date } | null = null;
-    for (const workspaceId of workspaceIds) {
+    for (const room of rooms) {
-      const row = await this.withTenant(workspaceId, async (c) => {
+      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",
+          "SELECT sync_cursor, updated_at FROM archive_sync_state WHERE room_id = $1 AND workspace_id = $2",
-          [this.stateId(workspaceId), workspaceId],
+          [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<void> {
+  async saveCursor(rooms: readonly RoomCursorTarget[], cursor: string): Promise<void> {
-    for (const workspaceId of workspaceIds) {
+    for (const room of rooms) {
-      await this.withTenant(workspaceId, (c) => c.query(
+      await this.withTenant(room.workspaceId, (c) => c.query(
-        `INSERT INTO archive_sync_state (id, workspace_id, sync_cursor, updated_at)
+        `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()`,
+         ON CONFLICT (room_id) DO UPDATE SET sync_cursor = EXCLUDED.sync_cursor, updated_at = now()`,
-        [this.stateId(workspaceId), workspaceId, cursor],
+        [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<T>(workspaceId: string, fn: (client: PoolClient) => Promise<T>): Promise<T> {
     const client = await this.pool.connect();
     try {

apps/archive-bot/src/webdav.test.ts

@@ -1,5 +1,53 @@
-import { describe, expect, it } from "vitest";
+import { afterEach, describe, expect, it, vi } from "vitest";
-import { davPath, parsePropfind } from "./webdav.js";
+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<Response>): 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<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    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("<d:multistatus/>")).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<Uint8Array>({
+      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<Uint8Array>({
+      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);
+  });
+});

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<Uint8Array>, signal: AbortSignal): Promise<void> {
     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,
+      body: payload,
-      headers: { "content-type": "application/octet-stream" },
+      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<Uint8Array> | string; headers?: Record<string, string> } = {},
+    init: { body?: ReadableStream<Uint8Array> | Buffer | string; headers?: Record<string, string> } = {},
   ): Promise<Response> {
     const res = await fetch(new URL(davPath(this.user, segments), this.baseUrl), {
       method,

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

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:?}