Hold a linked device's history on the server, not Gryt's relay (GRYT-2055)

src/db/interfaces.ts

@@ -283,6 +283,8 @@ export interface ServerConfigRecord {
   spam_filter_sensitivity: SpamSensitivity;
   /** Whether members may use a video avatar or banner, where the worker can transcode one. */
   video_profiles_enabled: boolean;
+  /** Holds a linked device's sealed history for it (GRYT-2055). */
+  link_history_enabled: boolean;
   /** Whether messages here may show other Gryt servers' custom emoji. Off: they load from another host. */
   external_emojis_enabled: boolean;
   is_configured: boolean;

src/db/sqlite/connection.ts

@@ -93,6 +93,7 @@ function createSchema(d: DatabaseSync): void {
       spam_filter INTEGER NOT NULL DEFAULT 1,
       spam_sensitivity TEXT NOT NULL DEFAULT 'normal',
       video_profiles INTEGER NOT NULL DEFAULT 1,
+      link_history INTEGER NOT NULL DEFAULT 1,
       external_emojis INTEGER NOT NULL DEFAULT 0,
       is_configured INTEGER NOT NULL DEFAULT 0,
       created_at TEXT NOT NULL,
@@ -1031,6 +1032,10 @@ function runMigrations(d: DatabaseSync): void {
   if (!hasColumn(d, "server_config", "video_profiles")) {
     d.exec("ALTER TABLE server_config ADD COLUMN video_profiles INTEGER NOT NULL DEFAULT 1");
   }
+
+  if (!hasColumn(d, "server_config", "link_history")) {
+    d.exec("ALTER TABLE server_config ADD COLUMN link_history INTEGER NOT NULL DEFAULT 1");
+  }
   // Off by default: another server's emoji loads from that host, which then sees who read it (GRYT-1660).
   if (!hasColumn(d, "server_config", "external_emojis")) {
     d.exec("ALTER TABLE server_config ADD COLUMN external_emojis INTEGER NOT NULL DEFAULT 0");

src/db/sqlite/linkHistorySetting.test.ts

@@ -0,0 +1,34 @@
+import assert from "node:assert/strict";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { after, before, describe, it } from "node:test";
+
+import { initSqlite } from "./connection";
+import { createServerConfigIfNotExists, getServerConfig, updateServerConfig } from "./servers";
+
+let dir: string;
+
+before(async () => {
+  dir = mkdtempSync(join(tmpdir(), "gryt-link-history-"));
+  process.env.DATA_DIR = dir;
+  await initSqlite();
+  await createServerConfigIfNotExists({ displayName: "Somewhere" });
+});
+
+after(() => {
+  delete process.env.DATA_DIR;
+  rmSync(dir, { recursive: true, force: true });
+});
+
+describe("holding linked devices' history (GRYT-2055)", () => {
+  it("is on until the owner turns it off, and stays off", async () => {
+    assert.equal((await getServerConfig())?.link_history_enabled, true);
+    await updateServerConfig({ linkHistory: false });
+    assert.equal((await getServerConfig())?.link_history_enabled, false);
+    await updateServerConfig({ displayName: "Elsewhere" });
+    assert.equal((await getServerConfig())?.link_history_enabled, false, "another change leaves it alone");
+    await updateServerConfig({ linkHistory: true });
+    assert.equal((await getServerConfig())?.link_history_enabled, true);
+  });
+});

src/db/sqlite/servers.ts

@@ -78,6 +78,7 @@ function rowToConfig(r: Record<string, unknown>): ServerConfigRecord {
     spam_filter_enabled: (r.spam_filter as number) !== 0,
     spam_filter_sensitivity: normalizeSpamSensitivity(r.spam_sensitivity),
     video_profiles_enabled: (r.video_profiles as number) !== 0,
+    link_history_enabled: (r.link_history as number) !== 0,
     external_emojis_enabled: (r.external_emojis as number) === 1,
     is_configured: (r.is_configured as number) === 1,
     created_at: fromIso(r.created_at as string),
@@ -244,6 +245,7 @@ export async function updateServerConfig(patch: {
   spamFilter?: boolean;
   spamSensitivity?: SpamSensitivity;
   videoProfiles?: boolean;
+  linkHistory?: boolean;
   externalEmojis?: boolean;
   isConfigured?: boolean;
 }): Promise<ServerConfigRecord> {
@@ -275,6 +277,7 @@ export async function updateServerConfig(patch: {
     spamFilter: { col: "spam_filter", transform: (v) => v ? 1 : 0 },
     spamSensitivity: { col: "spam_sensitivity", transform: (v) => normalizeSpamSensitivity(v) },
     videoProfiles: { col: "video_profiles", transform: (v) => v ? 1 : 0 },
+    linkHistory: { col: "link_history", transform: (v) => v ? 1 : 0 },
     externalEmojis: { col: "external_emojis", transform: (v) => v ? 1 : 0 },
     isConfigured: { col: "is_configured", transform: (v) => v ? 1 : 0 },
   };

src/index.ts

@@ -42,6 +42,7 @@ import { oEmbedRouter } from "./routes/oembed";
 import { mediaMetadataRouter } from "./routes/mediaMetadata";
 import { webhooksRouter } from "./routes/webhooks";
 import { discordImportRouter } from "./routes/discordImport";
+import { clearLinkHistoryOnStart, linkHistoryRouter } from "./routes/linkHistory";
 import { resumeDiscordImports } from "./import/discord/runner";
 import { noFileStorageMedia, uploadPathMedia } from "./import/discord/media";
 import { parsesOwnJson } from "./routes/ownJsonParsers";
@@ -124,6 +125,7 @@ const storageReady = (async () => {
     return;
   }
   initStorage();
+  await clearLinkHistoryOnStart();
   const bucket = uploadsFolder();
   await ensureBucket(bucket);
   consola.success(`Uploads are stored in ${bucket} under DATA_DIR`);
@@ -222,6 +224,7 @@ app.get("/info", httpRateLimit("http:public", RL_HTTP_PUBLIC), async (_req, res)
 
   let displayName = process.env.SERVER_NAME || "Unknown Server";
   let description = process.env.SERVER_DESCRIPTION || "A Gryt server";
+  let linkHistoryOn = !uploadsDisabled();
   let lanOpen = false;
   // Falls back to the stricter answer if the config cannot be read, so a
   // database wobble never advertises a server as easier to get into than it is.
@@ -260,6 +263,7 @@ app.get("/info", httpRateLimit("http:public", RL_HTTP_PUBLIC), async (_req, res)
     if (cfg?.display_name) displayName = cfg.display_name;
     if (cfg?.description) description = cfg.description;
     if (cfg?.lan_open) lanOpen = true;
+    if (cfg?.link_history_enabled === false) linkHistoryOn = false;
     if (cfg?.join_policy) joinPolicy = cfg.join_policy;
 
     const storedServerId = (cfg as { server_id?: string | null } | null)?.server_id;
@@ -289,6 +293,8 @@ app.get("/info", httpRateLimit("http:public", RL_HTTP_PUBLIC), async (_req, res)
     // anybody tries. Neither field says more than a failed join would.
     identityTiers: getAcceptedIdentityTiers(),
     joinPolicy,
+    // A linked device's history goes through here rather than through Gryt's relay.
+    linkHistory: linkHistoryOn,
   });
 });
 
@@ -369,6 +375,7 @@ app.use("/api/oembed", httpRateLimit("http:outbound", RL_HTTP_OUTBOUND), oEmbedR
 app.use("/api/media/metadata", httpRateLimit("http:outbound", RL_HTTP_OUTBOUND), mediaMetadataRouter);
 app.use("/api/webhooks", webhooksRouter);
 app.use("/api/imports/discord", httpRateLimit("http:api", RL_HTTP_API), discordImportRouter);
+app.use("/api/link-history", httpRateLimit("http:link-history", RL_HTTP_CHAT_UPLOAD), linkHistoryRouter);
 
 app.use(apiErrorHandler);
 

src/routes/linkHistory.test.ts

@@ -0,0 +1,88 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+
+import { createLinkHistoryStore, LINK_HISTORY_LIMITS, linkHistoryRouter } from "./linkHistory";
+
+const ID = "a".repeat(26);
+const OTHER = "b".repeat(26);
+
+function setup(limits = LINK_HISTORY_LIMITS) {
+  const disk = new Map<string, Buffer>();
+  let clock = 1_000;
+  const store = createLinkHistoryStore(
+    {
+      put: async (key, body) => void disk.set(key, body),
+      get: async (key) => disk.get(key)!,
+      del: async (key) => void disk.delete(key),
+    },
+    limits,
+    () => clock,
+  );
+  return { store, disk, tick: (ms: number) => (clock += ms) };
+}
+
+describe("link history store", () => {
+  it("gives a chunk back to the user who left it, and nobody else", async () => {
+    const { store } = setup();
+    assert.equal(await store.put("alice", ID, 0, Buffer.from("sealed")), null);
+    assert.deepEqual(await store.get("alice", ID, 0), Buffer.from("sealed"));
+    assert.equal(await store.get("mallory", ID, 0), null);
+    assert.equal(await store.put("mallory", ID, 1, Buffer.from("x")), "not_found", "another user can't add to it");
+    assert.equal(await store.remove("mallory", ID), false);
+  });
+
+  it("refuses chunks over the size, users over their total, and a third transfer", async () => {
+    const limits = { ...LINK_HISTORY_LIMITS, chunkBytes: 10, userBytes: 15 };
+    const { store } = setup(limits);
+    assert.equal(await store.put("alice", ID, 0, Buffer.alloc(11)), "too_large");
+    assert.equal(await store.put("alice", ID, 0, Buffer.alloc(10)), null);
+    assert.equal(await store.put("alice", ID, 1, Buffer.alloc(6)), "too_large", "10 + 6 is over 15");
+    assert.equal(await store.put("alice", ID, 0, Buffer.alloc(10)), null, "replacing a chunk counts once");
+    assert.equal(await store.put("alice", OTHER, 0, Buffer.alloc(5)), null);
+    assert.equal(await store.put("alice", "c".repeat(26), 0, Buffer.alloc(1)), "too_many");
+    assert.equal(await store.put("bob", "c".repeat(26), 0, Buffer.alloc(10)), null, "the limits are per user");
+  });
+
+  it("refuses ids and slots it doesn't expect", async () => {
+    const { store } = setup();
+    assert.equal(await store.put("alice", "short", 0, Buffer.from("x")), "bad_request");
+    assert.equal(await store.put("alice", "../../etc/passwd/xxxxxxxxxxxx", 0, Buffer.from("x")), "bad_request");
+    assert.equal(await store.put("alice", ID, -1, Buffer.from("x")), "bad_request");
+    assert.equal(await store.put("alice", ID, 1.5, Buffer.from("x")), "bad_request");
+    assert.equal(await store.put("alice", ID, LINK_HISTORY_LIMITS.slots, Buffer.from("x")), "bad_request");
+    assert.equal(await store.put("alice", ID, 0, Buffer.alloc(0)), "too_large");
+  });
+
+  it("frees a chunk's room once the new device has it", async () => {
+    const limits = { ...LINK_HISTORY_LIMITS, userBytes: 10 };
+    const { store, disk } = setup(limits);
+    await store.put("alice", ID, 0, Buffer.alloc(10));
+    assert.equal(await store.removeChunk("mallory", ID, 0), false);
+    assert.equal(await store.removeChunk("alice", ID, 0), true);
+    assert.equal(disk.size, 0);
+    assert.equal(await store.put("alice", ID, 1, Buffer.alloc(10)), null, "its bytes no longer count");
+  });
+
+  it("deletes a transfer's files when it's removed or runs out of time", async () => {
+    const { store, disk, tick } = setup();
+    await store.put("alice", ID, 0, Buffer.from("a"));
+    await store.put("alice", ID, 1, Buffer.from("b"));
+    assert.equal(await store.remove("alice", ID), true);
+    assert.equal(disk.size, 0);
+
+    await store.put("alice", OTHER, 0, Buffer.from("c"));
+    tick(LINK_HISTORY_LIMITS.ttlMs);
+    assert.equal(await store.get("alice", OTHER, 0), null);
+    assert.equal(disk.size, 0, "expired chunks are gone from disk");
+  });
+});
+
+describe("link history routes", () => {
+  it("authenticates before anything reads a body", () => {
+    type Layer = { name?: string; route?: object };
+    const stack: Layer[] = linkHistoryRouter.stack;
+    const auth = stack.findIndex((l) => l.name === "requireBearerToken");
+    const firstRoute = stack.findIndex((l) => l.route);
+    assert.ok(auth >= 0 && auth < firstRoute, "requireBearerToken runs before every route");
+  });
+});