diff --git a/docker-compose.keycloak.yml b/docker-compose.keycloak.yml index ab39e8d..d425e8f 100644 --- a/docker-compose.keycloak.yml +++ b/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: diff --git a/identity/README.md b/identity/README.md index 875110d..fe60ceb 100644 --- a/identity/README.md +++ b/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": ""}`, 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//`. 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 ` @@ -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 diff --git a/identity/src/index.ts b/identity/src/index.ts index 06676e5..bbe6456 100644 --- a/identity/src/index.ts +++ b/identity/src/index.ts @@ -1,6 +1,6 @@ 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); @@ -8,6 +8,8 @@ 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({ diff --git a/identity/src/pairing/chunks.ts b/identity/src/pairing/chunks.ts new file mode 100644 index 0000000..592b322 --- /dev/null +++ b/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): Partial { + const limits: Partial = {}; + 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; + bytes: number; +} + +export interface ChunkStoreOptions { + limits?: Partial; + now?: () => number; + log?: (line: string) => void; + /** Free bytes on the disk under `dir`. Tests swap it for a number. */ + freeBytes?: (dir: string) => Promise; +} + +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 { + const s = await statfs(dir); + return s.bavail * s.bsize; +} + +/** Sealed chunk files under `//`, and in memory only what they cost. */ +export class ChunkStore { + readonly dir: string; + readonly limits: ChunkLimits; + private readonly held = new Map(); + private readonly uploads = new Map(); + private readonly pending = new Set>(); + private readonly now: () => number; + private readonly log: (line: string) => void; + private readonly freeBytes: (dir: string) => Promise; + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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(work: Promise): Promise { + this.pending.add(work); + return work.finally(() => this.pending.delete(work)); + } +} diff --git a/identity/src/pairing/relay.ts b/identity/src/pairing/relay.ts index ef8d95b..93bb336 100644 --- a/identity/src/pairing/relay.ts +++ b/identity/src/pairing/relay.ts @@ -1,5 +1,7 @@ import { createHash, randomBytes, timingSafeEqual } from "node:crypto"; +import type { ChunkStore } from "./chunks.js"; + /** The relay's half of pairing, in memory only. See docs/pairing-design.md in Gryt-chat/crypto. */ export const LIMITS = { openMs: 5 * 60_000, @@ -105,6 +107,7 @@ export interface RelayOptions { limits?: Partial; now?: () => number; log?: (line: string) => void; + chunks?: ChunkStore; } export class Relay { @@ -118,6 +121,7 @@ export class Relay { private readonly refusals = new Map(); private readonly now: () => number; private readonly log: (line: string) => void; + readonly chunks: ChunkStore | null; private storedBytes = 0; private lastFlush: number; private timer: NodeJS.Timeout | null = null; @@ -126,11 +130,12 @@ export class Relay { this.limits = { ...LIMITS, ...options.limits }; this.now = options.now ?? Date.now; this.log = options.log ?? ((line) => console.log(line)); + this.chunks = options.chunks ?? null; this.lastFlush = this.now(); } start(intervalMs = 5000): void { - this.timer ??= setInterval(() => this.sweep(), intervalMs); + this.timer ??= setInterval(() => void this.sweep(), intervalMs); this.timer.unref(); } @@ -272,14 +277,48 @@ export class Relay { this.end(session, "closed"); } - sweep(): void { + /** A: one sealed history chunk into slot `n`. Only after the reveal. */ + async putChunk(id: string, token: string | undefined, ip: string, slot: string, body: unknown): Promise { + const session = this.find(id); + if (this.auth(session, token) !== "a") throw this.refuse(403, "wrong_side"); + if (session.state === "open" || session.state === "claimed") throw this.refuse(409, "not_revealed"); + const store = this.store(); + const n = this.slot(slot, store); + const bytes = typeof body === "string" && body.length > 0 && B64U.test(body) ? Buffer.from(body, "base64url") : null; + if (!bytes || bytes.toString("base64url") !== body) throw this.refuse(400, "invalid_body"); + if (bytes.length > store.limits.chunkBytes) throw this.refuse(413, "too_large"); + await this.counted(store.put(session.id, n, bytes, ip)); + this.find(id); + } + + /** N: the chunk in slot `n` as base64url. */ + async getChunk(id: string, token: string | undefined, slot: string): Promise { + const session = this.find(id); + if (this.auth(session, token) !== "n") throw this.refuse(403, "wrong_side"); + const store = this.store(); + const body = await this.counted(store.get(session.id, this.slot(slot, store))); + if (!body) throw this.refuse(404, "not_found"); + return body.toString("base64url"); + } + + async deleteChunk(id: string, token: string | undefined, slot: string): Promise { + const session = this.find(id); + if (this.auth(session, token) !== "n") throw this.refuse(403, "wrong_side"); + const store = this.store(); + await store.delete(session.id, this.slot(slot, store)); + } + + sweep(): Promise { const t = this.now(); for (const session of this.sessions.values()) if (session.expiresAt <= t) this.end(session, "expired"); for (const [id, e] of this.ended) if (e.until <= t) this.ended.delete(id); for (const [ip, h] of this.opened) if (h.start + this.limits.sessionWindowMs <= t) this.opened.delete(ip); for (const [ip, h] of this.codeFailures) if (h.start + this.limits.codeFailureWindowMs <= t) this.codeFailures.delete(ip); for (const [ip, until] of this.blocked) if (until <= t) this.blocked.delete(ip); - if (t - this.lastFlush >= 60_000) { + const flush = t - this.lastFlush >= 60_000; + if (flush) { + const stored = this.chunks?.flushStats(); + if (stored?.count) this.log(`pairing: stored ${stored.count} chunks, ${(stored.bytes / 1048576).toFixed(1)} MiB`); if (this.refusals.size > 0) { const counts = [...this.refusals].map(([reason, n]) => `${reason}=${n}`).join(" "); this.log(`pairing: refused ${counts}`); @@ -287,6 +326,30 @@ export class Relay { } this.lastFlush = t; } + const swept = this.chunks?.sweep((id) => this.sessions.has(id), flush); + return (swept ?? Promise.resolve()).catch((e) => { + this.log(`pairing: the chunk sweep failed: ${(e as NodeJS.ErrnoException)?.code ?? "unknown"}`); + }); + } + + private store(): ChunkStore { + if (!this.chunks) throw this.refuse(404, "not_found"); + return this.chunks; + } + + private slot(value: string, store: ChunkStore): number { + const n = /^(0|[1-9]\d{0,5})$/.test(value) ? Number(value) : -1; + if (n < 0 || n > store.limits.maxSlot) throw this.refuse(400, "invalid_slot"); + return n; + } + + private async counted(work: Promise): Promise { + try { + return await work; + } catch (e) { + if (e instanceof RelayError) this.refusals.set(e.code, (this.refusals.get(e.code) ?? 0) + 1); + throw e; + } } private live(id: string | undefined): Session | undefined { @@ -334,6 +397,7 @@ export class Relay { this.sessions.delete(session.id); if (session.code) this.codes.delete(session.code); this.storedBytes -= session.bytes; + this.chunks?.drop(session.id); this.ended.set(session.id, { reason, until: this.now() + this.limits.tombstoneMs }); for (const wake of [...session.waiters]) wake(); this.log(`pairing: session ${reason} after ${seconds(this.now() - session.createdAt)}, ${session.state}`); diff --git a/identity/src/pairing/routes.ts b/identity/src/pairing/routes.ts index 18a5052..7f3dfd7 100644 --- a/identity/src/pairing/routes.ts +++ b/identity/src/pairing/routes.ts @@ -5,6 +5,7 @@ import { Hono, type Context } from "hono"; import { bodyLimit } from "hono/body-limit"; import type { ContentfulStatusCode } from "hono/utils/http-status"; +import { CHUNK_LIMITS } from "./chunks.js"; import { Relay, RelayError } from "./relay.js"; /** Addresses allowed to speak for the client through CF-Connecting-IP: the tunnel, nothing else. */ @@ -47,7 +48,11 @@ export function visitorLocation(header: (name: string) => string | undefined): s export function pairingRoutes(relay: Relay, trusted: (address: string) => boolean): Hono { const routes = new Hono(); - const maxBody = Math.ceil((relay.limits.envelopeBytes * 4) / 3) + 1024; + const base64 = (bytes: number) => Math.ceil((bytes * 4) / 3) + 1024; + const tooLarge = (c: Context) => c.json({ error: "too_large" }, 413); + const messageLimit = bodyLimit({ maxSize: base64(relay.limits.envelopeBytes), onError: tooLarge }); + const chunkBytes = relay.chunks?.limits.chunkBytes ?? CHUNK_LIMITS.chunkBytes; + const chunkLimit = bodyLimit({ maxSize: base64(chunkBytes), onError: tooLarge }); const client = (c: Context) => { const socket = getConnInfo(c).remote.address ?? ""; @@ -62,7 +67,7 @@ export function pairingRoutes(relay: Relay, trusted: (address: string) => boolea return body as Record; }; - routes.use("*", bodyLimit({ maxSize: maxBody, onError: (c) => c.json({ error: "too_large" }, 413) })); + routes.use("*", (c, next) => (/\/chunks\/[^/]*$/.test(c.req.path) ? chunkLimit : messageLimit)(c, next)); routes.onError((err, c) => { if (err instanceof RelayError) return c.json({ error: err.code }, err.status as ContentfulStatusCode); @@ -93,6 +98,21 @@ export function pairingRoutes(relay: Relay, trusted: (address: string) => boolea return c.json({ messages: await relay.poll(c.req.param("id"), bearer(c), after, wait * 1000) }); }); + routes.put("/sessions/:id/chunks/:n", async (c) => { + const { ip } = client(c); + await relay.putChunk(c.req.param("id"), bearer(c), ip, c.req.param("n"), (await json(c)).body); + return c.json({}, 201); + }); + + routes.get("/sessions/:id/chunks/:n", async (c) => { + return c.json({ body: await relay.getChunk(c.req.param("id"), bearer(c), c.req.param("n")) }); + }); + + routes.delete("/sessions/:id/chunks/:n", async (c) => { + await relay.deleteChunk(c.req.param("id"), bearer(c), c.req.param("n")); + return c.body(null, 204); + }); + routes.delete("/sessions/:id", (c) => { relay.close(c.req.param("id"), bearer(c)); return c.body(null, 204); diff --git a/identity/src/routes.ts b/identity/src/routes.ts index d1d5554..18bbca9 100644 --- a/identity/src/routes.ts +++ b/identity/src/routes.ts @@ -1,14 +1,20 @@ +import { join } from "node:path"; + import { Hono } from "hono"; import { cors } from "hono/cors"; import { getCAPublicJwk } from "./keys.js"; import { verifyKeycloakToken } from "./keycloak.js"; import { issueCertificate } from "./certificate.js"; +import { ChunkStore, chunkLimitsFromEnv } from "./pairing/chunks.js"; import { Relay } from "./pairing/relay.js"; import { pairingRoutes, trustedProxies } from "./pairing/routes.js"; export const app = new Hono(); -export const relay = new Relay(); +export const chunks = new ChunkStore(join(process.env.GRYT_IDENTITY_DATA_DIR || "./data", "pairing"), { + limits: chunkLimitsFromEnv(process.env), +}); +export const relay = new Relay({ chunks }); app.use("*", cors()); app.route("/api/v1/pairing", pairingRoutes(relay, trustedProxies(process.env.GRYT_PAIRING_TRUSTED_PROXIES))); diff --git a/identity/test/chunks.test.ts b/identity/test/chunks.test.ts new file mode 100644 index 0000000..4ec20ed --- /dev/null +++ b/identity/test/chunks.test.ts @@ -0,0 +1,275 @@ +import assert from "node:assert/strict"; +import { createHash, randomBytes } from "node:crypto"; +import { once } from "node:events"; +import { existsSync } from "node:fs"; +import { mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises"; +import type { AddressInfo } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, before, beforeEach, test } from "node:test"; + +import { serve } from "@hono/node-server"; +import { Hono } from "hono"; + +import { ChunkStore, chunkLimitsFromEnv, type ChunkLimits } from "../src/pairing/chunks.js"; +import { Relay } from "../src/pairing/relay.js"; +import { pairingRoutes, trustedProxies } from "../src/pairing/routes.js"; + +const GiB = 1024 ** 3; +let clock = 0; +let logs: string[] = []; +let free = 100 * GiB; +let dir = ""; +let store: ChunkStore; +let relay: Relay; +let base = ""; +const server = serve({ + port: 0, + hostname: "127.0.0.1", + fetch: (req, env) => { + const app = new Hono(); + app.route("/p", pairingRoutes(relay, trustedProxies("127.0.0.1"))); + return app.fetch(req, env); + }, +}); + +before(async () => { + if (!server.listening) await once(server, "listening"); + base = `http://127.0.0.1:${(server.address() as AddressInfo).port}/p`; +}); +after(async () => { + server.close(); + if (dir) await rm(dir, { recursive: true, force: true }); +}); +beforeEach(() => fresh()); + +async function fresh(limits: Partial = {}) { + clock = 1_000_000; + logs = []; + free = 100 * GiB; + if (dir) await rm(dir, { recursive: true, force: true }); + dir = join(await mkdtemp(join(tmpdir(), "gryt-chunks-")), "pairing"); + const log = (line: string) => logs.push(line); + store = new ChunkStore(dir, { limits, now: () => clock, log, freeBytes: async () => free }); + await store.wipe(); + relay = new Relay({ now: () => clock, log, chunks: store }); +} + +async function call(method: string, path: string, opts: { body?: unknown; token?: string; ip?: string } = {}) { + const headers: Record = { "content-type": "application/json" }; + if (opts.token) headers.authorization = `Bearer ${opts.token}`; + headers["cf-connecting-ip"] = opts.ip ?? "203.0.113.1"; + const res = await fetch(base + path, { method, headers, body: opts.body === undefined ? undefined : JSON.stringify(opts.body) }); + const text = await res.text(); + return { status: res.status, json: text ? JSON.parse(text) : null }; +} + +const key = () => randomBytes(32); +const commitOf = (pkN: Buffer) => createHash("sha256").update("gryt-pair-v1 commit").update(pkN).digest(); + +/** A session past the reveal, with both bearer tokens. */ +async function revealed(opts: { approve?: boolean } = {}) { + const pkN = key(); + const opened = await call("POST", "/sessions", { body: { commit: commitOf(pkN).toString("base64url") }, ip: `198.51.100.${randomBytes(1)[0]}` }); + const { id, token } = opened.json; + const claim = await call("POST", "/sessions/claim", { body: { id, pkA: key().toString("base64url") } }); + const aToken = claim.json.token as string; + assert.equal((await call("POST", `/sessions/${id}/messages`, { token, body: { type: "reveal", pkN: pkN.toString("base64url") } })).status, 201); + if (opts.approve) await call("POST", `/sessions/${id}/messages`, { token: aToken, body: { type: "sealed", body: "AAAA" } }); + return { id: id as string, n: token as string, a: aToken }; +} + +const put = (s: { id: string; a: string }, slot: number | string, bytes: Buffer, ip?: string) => + call("PUT", `/sessions/${s.id}/chunks/${slot}`, { token: s.a, body: { body: bytes.toString("base64url") }, ip }); +const get = (s: { id: string; n: string }, slot: number) => call("GET", `/sessions/${s.id}/chunks/${slot}`, { token: s.n }); +const del = (s: { id: string; n: string }, slot: number) => call("DELETE", `/sessions/${s.id}/chunks/${slot}`, { token: s.n }); +const onDisk = async (id: string) => (existsSync(join(dir, id)) ? (await readdir(join(dir, id))).sort() : []); + +test("A uploads, N fetches, and the chunk is gone once N deletes it or fetches it twice", async () => { + const s = await revealed(); + const first = randomBytes(300_000); + assert.deepEqual(await put(s, 0, first), { status: 201, json: {} }); + assert.ok((await readFile(join(dir, s.id, "0"))).equals(first), "the file holds the sealed bytes as sent"); + + const fetched = await get(s, 0); + assert.equal(fetched.status, 200); + assert.ok(Buffer.from(fetched.json.body, "base64url").equals(first)); + assert.equal((await del(s, 0)).status, 204); + assert.deepEqual(await onDisk(s.id), []); + assert.deepEqual(await get(s, 0), { status: 404, json: { error: "not_found" } }); + assert.equal((await del(s, 0)).status, 204); + + const second = randomBytes(1000); + await put(s, 1, second); + assert.equal((await get(s, 1)).status, 200); + assert.ok(Buffer.from((await get(s, 1)).json.body, "base64url").equals(second)); + assert.deepEqual(await onDisk(s.id), []); + assert.equal((await get(s, 1)).status, 404); + assert.equal(store.totalBytes, 0); + + const logged = logs.join("\n"); + for (const secret of [s.id, s.n, s.a, first.toString("base64url").slice(0, 40)]) assert.ok(!logged.includes(secret)); +}); + +test("a retry with the same bytes is fine, different bytes in a used slot are not", async () => { + const s = await revealed(); + const bytes = randomBytes(500); + assert.equal((await put(s, 3, bytes)).status, 201); + assert.equal((await put(s, 3, bytes)).status, 201); + assert.deepEqual(await put(s, 3, randomBytes(500)), { status: 409, json: { error: "exists" } }); + assert.equal(store.totalBytes, 4096); +}); + +test("only A uploads, only N fetches, and only after the reveal", async () => { + const pkN = key(); + const opened = await call("POST", "/sessions", { body: { commit: commitOf(pkN).toString("base64url") } }); + const claim = await call("POST", "/sessions/claim", { body: { id: opened.json.id, pkA: key().toString("base64url") } }); + const early = { id: opened.json.id, n: opened.json.token, a: claim.json.token }; + assert.deepEqual(await put(early, 0, randomBytes(10)), { status: 409, json: { error: "not_revealed" } }); + + const s = await revealed(); + assert.equal((await put({ id: s.id, a: s.n }, 0, randomBytes(10))).json.error, "wrong_side"); + await put(s, 0, randomBytes(10)); + assert.equal((await get({ id: s.id, n: s.a }, 0)).json.error, "wrong_side"); + assert.equal((await del({ id: s.id, n: s.a }, 0)).json.error, "wrong_side"); + assert.equal((await put(s, "01", randomBytes(10))).json.error, "invalid_slot"); + assert.equal((await put(s, 65_536, randomBytes(10))).json.error, "invalid_slot"); + assert.equal((await put(s, -1, randomBytes(10))).json.error, "invalid_slot"); + const padded = await call("PUT", `/sessions/${s.id}/chunks/1`, { token: s.a, body: { body: "AAA=" } }); + assert.equal(padded.json.error, "invalid_body"); + const loose = await call("PUT", `/sessions/${s.id}/chunks/1`, { token: s.a, body: { body: "AB" } }); + assert.equal(loose.json.error, "invalid_body"); + assert.equal((await call("PUT", `/sessions/${s.id}/chunks/1`, { token: s.a, body: {} })).json.error, "invalid_body"); +}); + +test("a token from one session can't touch another session's chunks", async () => { + const mine = await revealed(); + const theirs = await revealed(); + await put(theirs, 0, randomBytes(100)); + assert.deepEqual(await put({ id: theirs.id, a: mine.a }, 1, randomBytes(100)), { status: 401, json: { error: "unauthorized" } }); + assert.equal((await get({ id: theirs.id, n: mine.n }, 0)).status, 401); + assert.equal((await del({ id: theirs.id, n: mine.n }, 0)).status, 401); + assert.equal((await call("GET", `/sessions/${theirs.id}/chunks/0`)).status, 401); + assert.deepEqual(await onDisk(theirs.id), ["0"]); +}); + +test("chunks go with their session: closed, or expired an hour after the claim", async () => { + const closed = await revealed({ approve: true }); + await put(closed, 0, randomBytes(100)); + assert.equal((await call("DELETE", `/sessions/${closed.id}`, { token: closed.a })).status, 204); + await store.idle(); + assert.ok(!existsSync(join(dir, closed.id))); + assert.deepEqual(await get(closed, 0), { status: 410, json: { error: "closed" } }); + + const s = await revealed({ approve: true }); + await put(s, 0, randomBytes(100)); + await put(s, 1, randomBytes(100)); + clock += 60 * 60_000 - 1; + await relay.sweep(); + assert.deepEqual(await onDisk(s.id), ["0", "1"]); + clock += 1; + await relay.sweep(); + await store.idle(); + assert.ok(!existsSync(join(dir, s.id))); + assert.deepEqual(await get(s, 0), { status: 410, json: { error: "expired" } }); + assert.equal(store.totalBytes, 0); + assert.deepEqual(await readdir(dir), []); +}); + +test("the sweep also drops a chunk an hour after its upload on its own", async () => { + await fresh({ maxAgeMs: 10 * 60_000 }); + const s = await revealed({ approve: true }); + await put(s, 0, randomBytes(100)); + clock += 5 * 60_000; + await put(s, 1, randomBytes(100)); + clock += 5 * 60_000; + await relay.sweep(); + assert.deepEqual(await onDisk(s.id), ["1"]); + assert.equal((await get(s, 0)).status, 404); +}); + +test("256 MiB a session, counted in sealed bytes", async () => { + await fresh({ sessionBytes: 5000, chunkBytes: 4000 }); + const s = await revealed(); + assert.equal((await put(s, 0, randomBytes(3000))).status, 201); + assert.deepEqual(await put(s, 1, randomBytes(2001)), { status: 507, json: { error: "session_full" } }); + assert.equal((await put(s, 1, randomBytes(2000))).status, 201); + assert.deepEqual(await put(s, 2, randomBytes(4001)), { status: 413, json: { error: "too_large" } }); + const other = await revealed(); + assert.equal((await put(other, 0, randomBytes(3000))).status, 201); + assert.equal((await del(s, 0)).status, 204); + assert.equal((await put(s, 2, randomBytes(3000))).status, 201); +}); + +test("2 GiB in total, with every file counted at 4 KiB at least", async () => { + await fresh({ totalBytes: 3 * 4096 }); + const one = await revealed(); + const two = await revealed(); + await put(one, 0, randomBytes(10)); + await put(one, 1, randomBytes(10)); + await put(two, 0, randomBytes(10)); + assert.deepEqual(await put(two, 1, randomBytes(10)), { status: 507, json: { error: "full" } }); + assert.deepEqual(await onDisk(two.id), ["0"]); + await call("DELETE", `/sessions/${one.id}`, { token: one.n }); + assert.equal((await put(two, 1, randomBytes(10))).status, 201); +}); + +test("uploads stop while the disk has under 5 GiB free", async () => { + const s = await revealed(); + free = 5 * GiB + 1000; + assert.equal((await put(s, 0, randomBytes(1000))).status, 201); + assert.deepEqual(await put(s, 1, randomBytes(1001)), { status: 507, json: { error: "disk_low" } }); + assert.deepEqual(await onDisk(s.id), ["0"]); + assert.equal(store.totalBytes, 4096); + free = 6 * GiB; + assert.equal((await put(s, 1, randomBytes(1001))).status, 201); + await relay.sweep(); + clock += 60_000; + await relay.sweep(); + assert.match(logs.join("\n"), /refused disk_low=1/); + assert.match(logs.join("\n"), /stored 2 chunks, 0\.0 MiB/); +}); + +test("1 GiB of uploads a day from one address", async () => { + await fresh({ uploadBytesPerIp: 2500 }); + const s = await revealed(); + assert.equal((await put(s, 0, randomBytes(1500), "192.0.2.1")).status, 201); + assert.deepEqual(await put(s, 1, randomBytes(1500), "192.0.2.1"), { status: 429, json: { error: "rate_limited" } }); + assert.equal((await put(s, 1, randomBytes(1500), "192.0.2.2")).status, 201); +}); + +test("a restart wipes the chunks of sessions it no longer knows, and nothing else", async () => { + const s = await revealed({ approve: true }); + await put(s, 0, randomBytes(100)); + await put(s, 1, randomBytes(100)); + await writeFile(join(dir, "not-a-session.txt"), "keep me"); + + store = new ChunkStore(dir, { now: () => clock, freeBytes: async () => free }); + assert.equal(await store.wipe(), 1); + relay = new Relay({ now: () => clock, chunks: store }); + assert.deepEqual(await readdir(dir), ["not-a-session.txt"]); + assert.deepEqual(await get(s, 0), { status: 404, json: { error: "not_found" } }); + assert.deepEqual(await put(s, 2, randomBytes(100)), { status: 404, json: { error: "not_found" } }); +}); + +test("the four limits come from the environment in MiB", () => { + assert.deepEqual(chunkLimitsFromEnv({}), {}); + assert.deepEqual( + chunkLimitsFromEnv({ + GRYT_PAIRING_CHUNKS_SESSION_MIB: "128", + GRYT_PAIRING_CHUNKS_TOTAL_MIB: "1024", + GRYT_PAIRING_CHUNKS_MIN_FREE_MIB: "0", + GRYT_PAIRING_CHUNKS_PER_IP_MIB: " 512 ", + }), + { sessionBytes: 128 * 1024 ** 2, totalBytes: GiB, minFreeBytes: 0, uploadBytesPerIp: 512 * 1024 ** 2 }, + ); + assert.throws(() => chunkLimitsFromEnv({ GRYT_PAIRING_CHUNKS_TOTAL_MIB: "2GiB" }), /GRYT_PAIRING_CHUNKS_TOTAL_MIB/); +}); + +// Last, since the server hangs up on a body it won't read and the next fetch can hit that socket. +test("a chunk request body past 2 MiB in base64 is refused before it's parsed", async () => { + const s = await revealed(); + assert.equal((await put(s, 0, randomBytes(2 * 1024 * 1024))).status, 201); + const huge = await call("PUT", `/sessions/${s.id}/chunks/1`, { token: s.a, body: { body: "A".repeat(3_000_000) } }); + assert.deepEqual(huge, { status: 413, json: { error: "too_large" } }); +});