From 12fb40ee4686007862e436f9dc93a3c45363ca7d Mon Sep 17 00:00:00 2001 From: m1ngsama Date: Thu, 15 Oct 2026 17:09:22 +0800 Subject: [PATCH] feat(gateway): report TNT identity events to m1ngsama ID --- gateway/src/events.ts | 94 +++++++++++++++++++++++++++++++++++++ gateway/src/main.ts | 14 +++++- gateway/src/module.ts | 15 +++++- gateway/test/events.test.ts | 63 +++++++++++++++++++++++++ gateway/test/module.test.ts | 13 +++++ 5 files changed, 197 insertions(+), 2 deletions(-) create mode 100644 gateway/src/events.ts create mode 100644 gateway/test/events.test.ts diff --git a/gateway/src/events.ts b/gateway/src/events.ts new file mode 100644 index 0000000..b9ef426 --- /dev/null +++ b/gateway/src/events.ts @@ -0,0 +1,94 @@ +const FLUSH_MS = 30_000; +const BATCH = 50; +const MAX_PENDING = 1000; + +export type IdentityEvent = { + id: string; + handle: string; + type: `tnt.${string}`; + at: number; + detail?: Record; +}; + +export type ReporterOptions = { + issuer: string; + secret: string; + /** Pending events survive a restart here. */ + file: string; + fetch?: typeof fetch; +}; + +/** Reports terminal identity events to m1ngsama ID, keeping them until it accepts them. */ +export function createReporter(options: ReporterOptions) { + const request = options.fetch ?? fetch; + const subs = new Map(); + let pending: IdentityEvent[] = []; + let flushing: Promise | null = null; + + async function load() { + const file = Bun.file(options.file); + if (!(await file.exists())) return; + try { + pending = JSON.parse(await file.text()) as IdentityEvent[]; + } catch { + pending = []; + } + } + + const save = () => Bun.write(options.file, JSON.stringify(pending)); + + async function subFor(handle: string): Promise { + const known = subs.get(handle); + if (known) return known; + const res = await request(`${options.issuer}/people/${encodeURIComponent(handle)}`); + if (res.status === 404) return null; + if (!res.ok) throw new Error(`people lookup answered ${res.status}`); + const { sub } = (await res.json()) as { sub?: unknown }; + if (typeof sub !== "string") return null; + subs.set(handle, sub); + return sub; + } + + async function send(): Promise { + while (pending.length > 0) { + const batch = pending.slice(0, BATCH); + const events = []; + const unknown = new Set(); + for (const event of batch) { + const sub = await subFor(event.handle); + if (sub === null) unknown.add(event.id); + else events.push({ id: event.id, sub, type: event.type, at: event.at, actor: { kind: "self" }, detail: event.detail }); + } + if (events.length > 0) { + const res = await request(`${options.issuer}/internal/events`, { + method: "POST", + headers: { authorization: `Basic ${btoa(`tnt:${options.secret}`)}`, "content-type": "application/json" }, + body: JSON.stringify({ events }), + }); + if (res.status !== 202) throw new Error(`events answered ${res.status}`); + } + // A handle with no person behind it has nobody to attribute the event to. + pending = pending.slice(batch.length); + await save(); + if (unknown.size > 0) console.error(`identity events: dropped ${unknown.size} for handles m1ngsama ID does not know`); + } + } + + function flush(): Promise { + flushing ??= send() + .catch((error: unknown) => console.error(`identity events: ${pending.length} waiting, ${String(error)}`)) + .finally(() => { + flushing = null; + }); + return flushing; + } + + async function report(event: Omit) { + pending.push({ id: crypto.randomUUID(), at: Date.now(), ...event }); + if (pending.length > MAX_PENDING) pending = pending.slice(-MAX_PENDING); + await save(); + void flush(); + } + + return { load, report, flush, start: () => setInterval(() => void flush(), FLUSH_MS), pending: () => pending }; +} diff --git a/gateway/src/main.ts b/gateway/src/main.ts index 6f0673f..c46c58d 100644 --- a/gateway/src/main.ts +++ b/gateway/src/main.ts @@ -1,5 +1,6 @@ import { createVerifier, remoteKeys } from "./auth"; import { createGateway } from "./gateway"; +import { createReporter } from "./events"; import { createGuard } from "./guard"; import { Ledger, readHistory } from "./history"; import { signIn } from "./identity"; @@ -22,6 +23,12 @@ const verify = createVerifier(issuer, remoteKeys(env.TNT_GATEWAY_JWKS_URL || `${ let identity = false; let direct = false; const signingIn = new Set(); +const eventsSecret = env.TNT_GATEWAY_EVENTS_SECRET; +const reporter = eventsSecret ? createReporter({ issuer, secret: eventsSecret, file: `${stateDir}/gateway-events.json` }) : null; +if (reporter) { + await reporter.load(); + reporter.start(); +} const guard = createGuard({ issuer, reserve: (handle) => { @@ -62,6 +69,8 @@ for await (const line of console) { direct = supportsPrivate(event.version); } else if (event.type === "private.created") { gateway.handlePrivate(event); + } else if (event.type === "identity.changed") { + void reporter?.report({ handle: event.handle, type: `tnt.${event.action}`, detail: event.detail }); } else if (event.type === "identity.requested") { const { session } = event; if (!identity || signingIn.has(session)) continue; @@ -70,7 +79,10 @@ for await (const line of console) { issuer, verify, notice: (plain_text) => write(encodeRecord({ type: "identity.notice", session, plain_text })), - verified: (handle) => write(encodeRecord({ type: "identity.verified", session, handle })), + verified: (handle) => { + write(encodeRecord({ type: "identity.verified", session, handle })); + void reporter?.report({ handle, type: "tnt.signed_in", detail: { method: "device_code" } }); + }, }).finally(() => signingIn.delete(session)); } else { if (identity && event.type === "presence.joined") void guard.check(event.nickname); diff --git a/gateway/src/module.ts b/gateway/src/module.ts index fc35ddd..0003280 100644 --- a/gateway/src/module.ts +++ b/gateway/src/module.ts @@ -2,11 +2,13 @@ export type TntEvent = | { type: "handshake"; version: string } | { type: "message.created"; message: { timestamp: string; sender: string; plain_text: string; verified: boolean } } | { type: "identity.requested"; session: string } + | { type: "identity.changed"; action: IdentityAction; handle: string; detail: Record } | { type: "private.created"; from: string; to: string; plain_text: string; timestamp: string; verified: boolean } | { type: "presence.snapshot"; nicknames: string[] } | { type: "presence.joined" | "presence.left"; nickname: string }; -export type RoomEvent = Exclude; +export type IdentityAction = "key_recognized" | "key_bound" | "signed_out"; +export type RoomEvent = Exclude; export type PrivateEvent = Extract; export const HANDSHAKE_OK = `${JSON.stringify({ @@ -68,6 +70,17 @@ export function parseTntLine(line: string): { reply: string | null; event: TntEv ? { reply: null, event: { type: "private.created", from, to, plain_text, timestamp, verified: record.verified === true } } : NOTHING; } + case "identity.recognized": + case "identity.key_bound": + case "identity.signed_out": { + if (!isString(record.handle)) return NOTHING; + const action = ({ "identity.recognized": "key_recognized", "identity.key_bound": "key_bound", "identity.signed_out": "signed_out" } as const)[record.type]; + const detail: Record = {}; + if (isString(record.fingerprint)) detail.fingerprint = record.fingerprint; + if (isString(record.previous_handle)) detail.previous_handle = record.previous_handle; + if (typeof record.forgot_key === "boolean") detail.forgot_key = record.forgot_key; + return { reply: null, event: { type: "identity.changed", action, handle: record.handle, detail } }; + } case "identity.requested": return isString(record.session) ? { reply: null, event: { type: "identity.requested", session: record.session } } : NOTHING; case "presence.snapshot": diff --git a/gateway/test/events.test.ts b/gateway/test/events.test.ts new file mode 100644 index 0000000..7fe3b8b --- /dev/null +++ b/gateway/test/events.test.ts @@ -0,0 +1,63 @@ +import { afterEach, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createReporter } from "../src/events"; + +let dir = ""; +afterEach(() => rmSync(dir, { recursive: true, force: true })); + +function setup(eventsStatus: () => number) { + dir = mkdtempSync(join(tmpdir(), "tnt-events-")); + const posted: unknown[] = []; + const lookups: string[] = []; + const fetch = (async (url: string, init?: RequestInit) => { + const path = new URL(url).pathname; + if (path.startsWith("/people/")) { + lookups.push(path); + return path === "/people/lune" ? Response.json({ sub: "sub-lune" }) : Response.json({ error: "not_found" }, { status: 404 }); + } + posted.push({ authorization: new Headers(init?.headers).get("authorization"), body: JSON.parse(String(init?.body)) }); + return new Response(null, { status: eventsStatus() }); + }) as unknown as typeof globalThis.fetch; + const options = { issuer: "https://id.test", secret: "s3cret", file: join(dir, "events.json"), fetch }; + return { reporter: createReporter(options), options, posted, lookups }; +} + +test("reports an event with the person's sub and forgets it once accepted", async () => { + const { reporter, posted, lookups } = setup(() => 202); + await reporter.report({ handle: "lune", type: "tnt.key_recognized", detail: { fingerprint: "SHA256:abc" } }); + await reporter.flush(); + await reporter.report({ handle: "lune", type: "tnt.signed_out", detail: { forgot_key: false } }); + await reporter.flush(); + expect(lookups).toEqual(["/people/lune"]); + expect(posted[0]).toMatchObject({ + authorization: `Basic ${btoa("tnt:s3cret")}`, + body: { events: [{ sub: "sub-lune", type: "tnt.key_recognized", actor: { kind: "self" }, detail: { fingerprint: "SHA256:abc" } }] }, + }); + expect(reporter.pending()).toEqual([]); +}); + +test("keeps refused events on disk and sends them after a restart", async () => { + let status = 503; + const { reporter, options, posted } = setup(() => status); + await reporter.report({ handle: "lune", type: "tnt.signed_in", detail: { method: "device_code" } }); + await reporter.flush(); + expect(reporter.pending()).toHaveLength(1); + const id = reporter.pending()[0]!.id; + + status = 202; + const restarted = createReporter(options); + await restarted.load(); + await restarted.flush(); + expect(restarted.pending()).toEqual([]); + expect((posted.at(-1) as { body: { events: { id: string }[] } }).body.events[0]!.id).toBe(id); +}); + +test("drops events for a handle m1ngsama ID does not know", async () => { + const { reporter, posted } = setup(() => 202); + await reporter.report({ handle: "ghost", type: "tnt.signed_out" }); + await reporter.flush(); + expect(posted).toEqual([]); + expect(reporter.pending()).toEqual([]); +}); diff --git a/gateway/test/module.test.ts b/gateway/test/module.test.ts index 613ff14..6b8cb70 100644 --- a/gateway/test/module.test.ts +++ b/gateway/test/module.test.ts @@ -49,6 +49,19 @@ describe("parseTntLine", () => { expect(supportsPrivate("1.6.0")).toBe(true); }); + test("turns TNT's identity notifications into events", () => { + expect(parseTntLine('{"type":"identity.recognized","session":"s-8","handle":"lune","fingerprint":"SHA256:a"}').event).toEqual({ + type: "identity.changed", action: "key_recognized", handle: "lune", detail: { fingerprint: "SHA256:a" }, + }); + expect(parseTntLine('{"type":"identity.key_bound","session":"s-9","handle":"lune","fingerprint":"SHA256:a","previous_handle":"zed"}').event).toEqual({ + type: "identity.changed", action: "key_bound", handle: "lune", detail: { fingerprint: "SHA256:a", previous_handle: "zed" }, + }); + expect(parseTntLine('{"type":"identity.signed_out","session":"s-2","handle":"lune","forgot_key":false}')).toEqual({ + reply: null, event: { type: "identity.changed", action: "signed_out", handle: "lune", detail: { forgot_key: false } }, + }); + expect(parseTntLine('{"type":"identity.signed_out","session":"s-2"}')).toEqual({ reply: null, event: null }); + }); + test("parses presence records without answering them", () => { expect(parseTntLine('{"type":"presence.snapshot","nicknames":["a",1,"b"]}')).toEqual({ reply: null,