Store pairing history chunks on the identity service's disk (GRYT-1591)

docker-compose.keycloak.yml

@@ -459,6 +459,11 @@ services:
       GRYT_IDENTITY_DATA_DIR: /app/data
       GRYT_CERT_LIFETIME_DAYS: ${GRYT_CERT_LIFETIME_DAYS:-30}
       GRYT_PAIRING_TRUSTED_PROXIES: ${GRYT_PAIRING_TRUSTED_PROXIES:-}
+      # History chunks go under /app/data/pairing, on the same disk as the Keycloak database.
+      GRYT_PAIRING_CHUNKS_SESSION_MIB: ${GRYT_PAIRING_CHUNKS_SESSION_MIB:-256}
+      GRYT_PAIRING_CHUNKS_TOTAL_MIB: ${GRYT_PAIRING_CHUNKS_TOTAL_MIB:-2048}
+      GRYT_PAIRING_CHUNKS_MIN_FREE_MIB: ${GRYT_PAIRING_CHUNKS_MIN_FREE_MIB:-5120}
+      GRYT_PAIRING_CHUNKS_PER_IP_MIB: ${GRYT_PAIRING_CHUNKS_PER_IP_MIB:-1024}
     ports:
       - "${GRYT_IDENTITY_BIND:-0.0.0.0}:${GRYT_IDENTITY_PORT:-18081}:3000"
     volumes:

identity/README.md

@@ -16,6 +16,9 @@ Servers and other clients verify these certificates against the public JWKS endp
 | `POST` | `/api/v1/pairing/sessions/:id/messages` | Send the reveal or a sealed message |
 | `GET` | `/api/v1/pairing/sessions/:id/messages` | Read the other side's messages, long-polling up to 25 seconds |
 | `DELETE` | `/api/v1/pairing/sessions/:id` | Cancel or finish |
+| `PUT` | `/api/v1/pairing/sessions/:id/chunks/:n` | Store a sealed history chunk in slot `n` (the approving device) |
+| `GET` | `/api/v1/pairing/sessions/:id/chunks/:n` | Fetch one (the new device) |
+| `DELETE` | `/api/v1/pairing/sessions/:id/chunks/:n` | Done with it (the new device) |
 
 ### Pairing relay
 
@@ -24,7 +27,7 @@ and can't open them. Nothing needs a Keycloak token, so guests use it too. The p
 [`docs/pairing-design.md`](https://github.com/Gryt-chat/crypto/blob/main/docs/pairing-design.md)
 in Gryt-chat/crypto.
 
-Everything is in memory, so a restart drops every session and people start again. Keys,
+Sessions and messages live in memory, so a restart drops every session and people start again. Keys,
 commitments and sealed bodies are unpadded base64url. The logs hold counts and durations, and
 never an IP, a code, a session id or a location.
 
@@ -35,6 +38,31 @@ address, so ten sessions in ten minutes would be the limit for everybody at once
 `CF-IPCountry` and `CF-IPCity` headers are read from the same trusted addresses, and turned
 into "Oslo, Norway" for the approving device.
 
+#### History chunks
+
+Message history is the one thing that goes on disk. The approving device uploads it in sealed
+chunks of up to 2 MiB, as `{"body": "<base64url>"}`, and the new device fetches them. The relay
+can't open them and never logs them. Each one is saved as it arrived, under
+`$GRYT_IDENTITY_DATA_DIR/pairing/<session>/<n>`. In compose that's the `gryt-identity-data`
+volume.
+
+A chunk goes when the new device deletes it or fetches it a second time, when its session
+closes or expires, or an hour after upload. The folder also gets emptied at startup, because
+the sessions didn't survive the restart. Only folders named like a session id get removed, so
+the CA key in the same volume is left alone.
+
+That volume sits on the same disk as Keycloak's Postgres, and a full disk takes the auth
+database down. So uploads stop with `507` when any of these is hit:
+
+| Limit | Default | Error |
+|---|---|---|
+| Chunks in one session | 256 MiB | `session_full` |
+| Chunks in total, with each file counted as 4 KiB at least | 2 GiB | `full` |
+| Free space on the volume after the upload | 5 GiB | `disk_low` |
+
+On top of that, one address can upload 1 GiB a day, then gets `429 rate_limited`. The apps
+treat a `507` as the end of what fits, and mark the history as cut short.
+
 ### `POST /api/v1/certificate`
 
 **Headers:** `Authorization: Bearer <keycloak-access-token>`
@@ -65,6 +93,10 @@ into "Oslo, Norway" for the approving device.
 | `GRYT_IDENTITY_DATA_DIR` | Directory for auto-generated CA key storage | `./data` |
 | `GRYT_CERT_LIFETIME_DAYS` | Certificate validity period in days | `30` |
 | `GRYT_PAIRING_TRUSTED_PROXIES` | Comma-separated addresses or CIDR ranges allowed to set `CF-Connecting-IP` | _(none)_ |
+| `GRYT_PAIRING_CHUNKS_SESSION_MIB` | History chunks one pairing can store, in MiB | `256` |
+| `GRYT_PAIRING_CHUNKS_TOTAL_MIB` | History chunks stored across every pairing, in MiB | `2048` |
+| `GRYT_PAIRING_CHUNKS_MIN_FREE_MIB` | Free space the volume keeps; uploads stop below it | `5120` |
+| `GRYT_PAIRING_CHUNKS_PER_IP_MIB` | History chunks one address can upload a day, in MiB | `1024` |
 
 ## Stack
 

identity/src/index.ts

@@ -1,13 +1,15 @@
 import { serve } from "@hono/node-server";
 
-import { app, relay } from "./routes.js";
+import { app, chunks, relay } from "./routes.js";
 
 const port = parseInt(process.env.PORT || "3000", 10);
 
 console.log(`Gryt Identity Service starting on port ${port}...`);
 if (!process.env.GRYT_PAIRING_TRUSTED_PROXIES) {
   console.warn("pairing: GRYT_PAIRING_TRUSTED_PROXIES is empty, so every client counts as the tunnel's address");
 }
+const wiped = await chunks.wipe();
+if (wiped > 0) console.log(`pairing: removed ${wiped} chunk folders left from before the restart`);
 relay.start();
 
 serve({

identity/src/pairing/chunks.ts

@@ -0,0 +1,258 @@
+import { createHash } from "node:crypto";
+import { mkdir, readFile, readdir, rm, statfs, unlink, writeFile } from "node:fs/promises";
+import { join } from "node:path";
+
+import { RelayError } from "./relay.js";
+
+const MiB = 1024 * 1024;
+
+/** History chunk storage on disk. See "The disk" in docs/pairing-design.md in Gryt-chat/crypto. */
+export const CHUNK_LIMITS = {
+  chunkBytes: 2 * MiB,
+  sessionBytes: 256 * MiB,
+  totalBytes: 2048 * MiB,
+  minFreeBytes: 5120 * MiB,
+  uploadBytesPerIp: 1024 * MiB,
+  uploadWindowMs: 24 * 60 * 60_000,
+  maxAgeMs: 60 * 60_000,
+  maxSlot: 65_535,
+  // What the total cap charges a file at least, so a flood of tiny chunks still counts.
+  fileBytes: 4096,
+};
+export type ChunkLimits = typeof CHUNK_LIMITS;
+
+const ENV: [string, keyof ChunkLimits][] = [
+  ["GRYT_PAIRING_CHUNKS_SESSION_MIB", "sessionBytes"],
+  ["GRYT_PAIRING_CHUNKS_TOTAL_MIB", "totalBytes"],
+  ["GRYT_PAIRING_CHUNKS_MIN_FREE_MIB", "minFreeBytes"],
+  ["GRYT_PAIRING_CHUNKS_PER_IP_MIB", "uploadBytesPerIp"],
+];
+
+export function chunkLimitsFromEnv(env: Record<string, string | undefined>): Partial<ChunkLimits> {
+  const limits: Partial<ChunkLimits> = {};
+  for (const [name, key] of ENV) {
+    const raw = env[name]?.trim();
+    if (!raw) continue;
+    if (!/^\d{1,7}$/.test(raw)) throw new Error(`${name}: "${raw}" isn't a whole number of MiB.`);
+    limits[key] = Number(raw) * MiB;
+  }
+  return limits;
+}
+
+// Session ids are 16 random bytes in Crockford base32. Nothing else in the folder is ours to delete.
+const SESSION_DIR = /^[0-9A-HJKMNP-TV-Z]{26}$/;
+
+interface Slot {
+  bytes: number;
+  hash: Buffer;
+  ready: boolean;
+  storedAt: number;
+  fetches: number;
+}
+
+interface Held {
+  slots: Map<number, Slot>;
+  bytes: number;
+}
+
+export interface ChunkStoreOptions {
+  limits?: Partial<ChunkLimits>;
+  now?: () => number;
+  log?: (line: string) => void;
+  /** Free bytes on the disk under `dir`. Tests swap it for a number. */
+  freeBytes?: (dir: string) => Promise<number>;
+}
+
+const sha256 = (data: Buffer) => createHash("sha256").update(data).digest();
+const errorCode = (e: unknown) => (e as NodeJS.ErrnoException)?.code ?? "unknown";
+
+async function statfsFree(dir: string): Promise<number> {
+  const s = await statfs(dir);
+  return s.bavail * s.bsize;
+}
+
+/** Sealed chunk files under `<dir>/<session>/<n>`, and in memory only what they cost. */
+export class ChunkStore {
+  readonly dir: string;
+  readonly limits: ChunkLimits;
+  private readonly held = new Map<string, Held>();
+  private readonly uploads = new Map<string, { bytes: number; start: number }>();
+  private readonly pending = new Set<Promise<unknown>>();
+  private readonly now: () => number;
+  private readonly log: (line: string) => void;
+  private readonly freeBytes: (dir: string) => Promise<number>;
+  private total = 0;
+  private stored = { count: 0, bytes: 0 };
+
+  constructor(dir: string, options: ChunkStoreOptions = {}) {
+    this.dir = dir;
+    this.limits = { ...CHUNK_LIMITS, ...options.limits };
+    this.now = options.now ?? Date.now;
+    this.log = options.log ?? ((line) => console.log(line));
+    this.freeBytes = options.freeBytes ?? statfsFree;
+  }
+
+  /** At startup: the sessions went with the last process, so their chunks go too. */
+  async wipe(): Promise<number> {
+    await mkdir(this.dir, { recursive: true, mode: 0o700 });
+    let removed = 0;
+    for (const name of await readdir(this.dir)) {
+      if (!SESSION_DIR.test(name)) continue;
+      await rm(join(this.dir, name), { recursive: true, force: true }).then(
+        () => removed++,
+        (e) => this.log(`pairing: couldn't remove a chunk folder at startup: ${errorCode(e)}`),
+      );
+    }
+    return removed;
+  }
+
+  /** What the total cap currently counts, in bytes. */
+  get totalBytes(): number {
+    return this.total;
+  }
+
+  async put(id: string, n: number, body: Buffer, ip: string): Promise<void> {
+    let held = this.held.get(id);
+    const hash = sha256(body);
+    const existing = held?.slots.get(n);
+    if (existing) {
+      // A retry after a lost response sends the same bytes again, and that's fine.
+      if (existing.ready && existing.hash.equals(hash)) return;
+      throw new RelayError(409, "exists");
+    }
+    const charge = Math.max(body.length, this.limits.fileBytes);
+    if ((held?.bytes ?? 0) + body.length > this.limits.sessionBytes) throw new RelayError(507, "session_full");
+    if (this.total + charge > this.limits.totalBytes) throw new RelayError(507, "full");
+    const t = this.now();
+    let up = this.uploads.get(ip);
+    if (!up || up.start + this.limits.uploadWindowMs <= t) {
+      up = { bytes: 0, start: t };
+      this.uploads.set(ip, up);
+    }
+    if (up.bytes + body.length > this.limits.uploadBytesPerIp) throw new RelayError(429, "rate_limited");
+
+    if (!held) {
+      held = { slots: new Map(), bytes: 0 };
+      this.held.set(id, held);
+    }
+    const slot: Slot = { bytes: body.length, hash, ready: false, storedAt: t, fetches: 0 };
+    held.slots.set(n, slot);
+    held.bytes += body.length;
+    this.total += charge;
+    up.bytes += body.length;
+
+    const file = join(this.dir, id, String(n));
+    try {
+      if ((await this.freeBytes(this.dir)) - body.length < this.limits.minFreeBytes) {
+        throw new RelayError(507, "disk_low");
+      }
+      await mkdir(join(this.dir, id), { recursive: true, mode: 0o700 });
+      await writeFile(file, body, { flag: "wx", mode: 0o600 });
+    } catch (e) {
+      this.forget(id, n, slot);
+      up.bytes = Math.max(0, up.bytes - body.length);
+      if (e instanceof RelayError) throw e;
+      this.log(`pairing: storing a chunk failed: ${errorCode(e)}`);
+      throw new RelayError(503, "busy");
+    }
+    if (this.held.get(id)?.slots.get(n) !== slot) {
+      await this.track(rm(file, { force: true }).catch(() => undefined));
+      return;
+    }
+    slot.ready = true;
+    slot.storedAt = this.now();
+    this.stored.count++;
+    this.stored.bytes += body.length;
+  }
+
+  /** The chunk in slot `n`, or null. The second fetch deletes it, as N's DELETE would. */
+  async get(id: string, n: number): Promise<Buffer | null> {
+    const slot = this.held.get(id)?.slots.get(n);
+    if (!slot?.ready) return null;
+    let body: Buffer;
+    try {
+      body = await readFile(join(this.dir, id, String(n)));
+    } catch (e) {
+      if (errorCode(e) !== "ENOENT") this.log(`pairing: reading a chunk failed: ${errorCode(e)}`);
+      this.forget(id, n, slot);
+      if (errorCode(e) === "ENOENT") return null;
+      throw new RelayError(503, "busy");
+    }
+    if (++slot.fetches >= 2) await this.delete(id, n);
+    return body;
+  }
+
+  async delete(id: string, n: number): Promise<void> {
+    const slot = this.held.get(id)?.slots.get(n);
+    if (!slot) return;
+    this.forget(id, n, slot);
+    if (!slot.ready) return;
+    await this.track(
+      unlink(join(this.dir, id, String(n))).catch((e) => {
+        if (errorCode(e) !== "ENOENT") this.log(`pairing: deleting a chunk failed: ${errorCode(e)}`);
+      }),
+    );
+  }
+
+  /** The session is gone: every chunk it had goes with it. */
+  drop(id: string): void {
+    const held = this.held.get(id);
+    if (held) for (const [n, slot] of held.slots) this.forget(id, n, slot);
+    this.held.delete(id);
+    void this.track(
+      rm(join(this.dir, id), { recursive: true, force: true }).catch((e) => {
+        this.log(`pairing: removing a session's chunks failed: ${errorCode(e)}`);
+      }),
+    );
+  }
+
+  /** Chunks past their hour, sessions the relay no longer has, and folders nobody owns. */
+  async sweep(live: (id: string) => boolean, orphans = false): Promise<void> {
+    const t = this.now();
+    for (const [id, held] of this.held) {
+      if (!live(id)) {
+        this.drop(id);
+        continue;
+      }
+      for (const [n, slot] of held.slots) {
+        if (slot.ready && slot.storedAt + this.limits.maxAgeMs <= t) await this.delete(id, n);
+      }
+    }
+    for (const [ip, up] of this.uploads) if (up.start + this.limits.uploadWindowMs <= t) this.uploads.delete(ip);
+    if (!orphans) return;
+    let names: string[];
+    try {
+      names = await readdir(this.dir);
+    } catch (e) {
+      return this.log(`pairing: listing the chunk folder failed: ${errorCode(e)}`);
+    }
+    for (const name of names) {
+      if (SESSION_DIR.test(name) && !this.held.has(name) && !live(name)) this.drop(name);
+    }
+  }
+
+  /** Chunks stored since the last call, for the relay's once-a-minute log line. */
+  flushStats(): { count: number; bytes: number } {
+    const stats = this.stored;
+    this.stored = { count: 0, bytes: 0 };
+    return stats;
+  }
+
+  /** Resolves once every file operation started so far has finished. */
+  async idle(): Promise<void> {
+    while (this.pending.size > 0) await Promise.all([...this.pending]);
+  }
+
+  private forget(id: string, n: number, slot: Slot): void {
+    const held = this.held.get(id);
+    if (!held || held.slots.get(n) !== slot) return;
+    held.slots.delete(n);
+    held.bytes -= slot.bytes;
+    this.total -= Math.max(slot.bytes, this.limits.fileBytes);
+  }
+
+  private track<T>(work: Promise<T>): Promise<T> {
+    this.pending.add(work);
+    return work.finally(() => this.pending.delete(work));
+  }
+}