v4: federated sign-in closes the holes a review found — a callback link made from someone else's sign-in signed the visitor in to that account, a password reset left the user's sessions on other instances open, and events could be lost, reordered or delayed. FederatedSignIn keeps the sign-in's state in sessionStorage (utils/auth/federation.ts) and FederationCallback finishes only a sign-in this tab started. A password reset tells the instances the user signed in to, as a change does. The outbox delivers each instance's events in the order they were made, the first that fails holding back the rest, so a late older update no longer undoes a newer one, and delivers to every instance at once, so a dead one delays none of the others; an event queued during a run is sent by it.
Receiving, a home's keys that can't be fetched answer 502, which the sender retries, instead of 401, which it dropped; a bad update answers 400 rather than 502; a blocked instance's events answer 403; events are limited to a burst of 120 per instance, and a token naming an unknown key refetches the keys at most once a minute. An avatar must be on its issuer's origin, and its URL is versioned by its file's write time instead of updatedAt, so a profile edit no longer makes clients download it again. An instance that doesn't answer can be blocked, keeping its name, though not trusted. 303 backend tests and 703 frontend tests pass, the federation tests 8 times in a row, and deno check, tsc, eslint, doc:lint and i18n:compile --strict are clean. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
khannurien
@@ -13,7 +13,7 @@
|
||||
* the instance key.
|
||||
* 3. While the user's grant lasts, the client hears of their changes: a new
|
||||
* name, email or avatar, and the end of their sessions after a password
|
||||
* change. {@link notifyClients} queues the signed events in
|
||||
* change or reset. {@link notifyClients} queues the signed events in
|
||||
* `federation_outbox`, and {@link outboxJob} delivers them.
|
||||
*
|
||||
* Only a local account signs in elsewhere: a user who came from another
|
||||
@@ -26,7 +26,12 @@
|
||||
import { createHash, randomBytes } from "node:crypto";
|
||||
import { APIException } from "../lib/api-exception.ts";
|
||||
import { db, transaction } from "../db/connection.ts";
|
||||
import { FRONTEND_URL, OG_SITE_NAME, PUBLIC_URL } from "../config.ts";
|
||||
import {
|
||||
AVATARS_DIR,
|
||||
FRONTEND_URL,
|
||||
OG_SITE_NAME,
|
||||
PUBLIC_URL,
|
||||
} from "../config.ts";
|
||||
import { signInstanceToken } from "../lib/jwt.ts";
|
||||
import type { Job } from "../lib/scheduler.ts";
|
||||
import {
|
||||
@@ -105,9 +110,23 @@ function sha256(value: string): string {
|
||||
return createHash("sha256").update(value).digest("base64url");
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the version of `user`'s avatar: when its file was last written,
|
||||
* which only a new avatar does, unlike `updatedAt`, which every profile edit
|
||||
* bumps.
|
||||
*/
|
||||
function avatarVersion(user: User): number {
|
||||
try {
|
||||
return Deno.statSync(`${AVATARS_DIR}/${user.id}`).mtime?.getTime() ?? 0;
|
||||
} catch {
|
||||
return user.updatedAt?.getTime() ?? 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns what this instance tells another of `user`: their name, email and
|
||||
* avatar, whose URL changes with each new one.
|
||||
* avatar, whose URL changes with each new one, so that a client fetches it
|
||||
* again only then.
|
||||
*/
|
||||
export function identityClaims(user: User): IdentityClaims {
|
||||
return {
|
||||
@@ -115,7 +134,7 @@ export function identityClaims(user: User): IdentityClaims {
|
||||
email: user.email,
|
||||
...(user.avatarMime && {
|
||||
picture: `${PUBLIC_URL}/api/users/${user.id}/avatar?v=${
|
||||
user.updatedAt?.getTime() ?? 0
|
||||
avatarVersion(user)
|
||||
}`,
|
||||
}),
|
||||
};
|
||||
@@ -348,7 +367,8 @@ export async function revokeGrant(userId: string, host: string): Promise<void> {
|
||||
* `clientIds`: `update` when their name, email or avatar changed, `revoke`
|
||||
* when their sessions there must end. The routes that change a user call it,
|
||||
* and for an `update`, `event` may be just the type: the claims are read now.
|
||||
* The events are queued in the outbox and their delivery starts at once.
|
||||
* The events are queued in the outbox and their delivery starts at once,
|
||||
* after those queued before for the same instance ({@link deliverTo}).
|
||||
*/
|
||||
export async function notifyClients(
|
||||
userId: string,
|
||||
@@ -381,52 +401,80 @@ export async function notifyClients(
|
||||
now.toISOString(),
|
||||
);
|
||||
}
|
||||
deliverOutbox().catch((err) =>
|
||||
console.error("[federation] delivery failed:", err)
|
||||
);
|
||||
for (const clientId of clients) deliverTo(clientId);
|
||||
}
|
||||
|
||||
/** Set while {@link deliverOutbox} runs, so that two runs don't overlap. */
|
||||
let delivering = false;
|
||||
/**
|
||||
* The instances whose events {@link deliverTo} is sending, so that two runs
|
||||
* don't send the same events, and those whose events came meanwhile, which
|
||||
* the running one sends too once done.
|
||||
*/
|
||||
const delivering = new Set<string>();
|
||||
const queuedMeanwhile = new Set<string>();
|
||||
|
||||
/**
|
||||
* Sends each due event of the outbox to its instance. One it took, or refused
|
||||
* for good, leaves the outbox; one that failed waits twice as long as the time
|
||||
* before, up to 6 hours, until it expires after 3 days.
|
||||
* Sends the outbox's events to each instance that has some, every instance at
|
||||
* once so that one that doesn't answer delays none of the others.
|
||||
*/
|
||||
export async function deliverOutbox(): Promise<void> {
|
||||
if (delivering) return;
|
||||
delivering = true;
|
||||
db.prepare(`DELETE FROM federation_outbox WHERE created_at < ?;`).run(
|
||||
new Date(Date.now() - EVENT_TTL_MS).toISOString(),
|
||||
);
|
||||
const clients = db.prepare(
|
||||
`SELECT DISTINCT client_id FROM federation_outbox;`,
|
||||
).all() as { client_id: string }[];
|
||||
await Promise.all(clients.map((c) => deliverTo(c.client_id)));
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends the events for the instance `clientId` in the order they were made,
|
||||
* as long as it takes them, since each `update` carries all the claims and a
|
||||
* late older one would undo a newer one. One it took, or refused for good,
|
||||
* leaves the outbox. The first that fails stops the run, and waits twice as
|
||||
* long as the time before, up to 6 hours, with the events behind it, until
|
||||
* it expires after 3 days.
|
||||
*/
|
||||
async function deliverTo(clientId: string): Promise<void> {
|
||||
if (delivering.has(clientId)) {
|
||||
queuedMeanwhile.add(clientId);
|
||||
return;
|
||||
}
|
||||
delivering.add(clientId);
|
||||
try {
|
||||
const now = new Date();
|
||||
db.prepare(`DELETE FROM federation_outbox WHERE created_at < ?;`).run(
|
||||
new Date(now.getTime() - EVENT_TTL_MS).toISOString(),
|
||||
);
|
||||
const due = db.prepare(
|
||||
`SELECT id, client_id, event, attempts FROM federation_outbox
|
||||
WHERE next_attempt_at <= ? ORDER BY created_at LIMIT 100;`,
|
||||
).all(now.toISOString()) as {
|
||||
id: string;
|
||||
client_id: string;
|
||||
event: string;
|
||||
attempts: number;
|
||||
}[];
|
||||
for (const row of due) {
|
||||
try {
|
||||
if (!await postEvent(row.client_id, row.event)) {
|
||||
console.error(`[federation] ${row.client_id} refused an event`);
|
||||
}
|
||||
db.prepare(`DELETE FROM federation_outbox WHERE id = ?;`).run(row.id);
|
||||
} catch (err) {
|
||||
const wait = Math.min(60_000 * 2 ** row.attempts, MAX_RETRY_MS);
|
||||
db.prepare(
|
||||
`UPDATE federation_outbox SET attempts = attempts + 1, next_attempt_at = ? WHERE id = ?;`,
|
||||
).run(new Date(Date.now() + wait).toISOString(), row.id);
|
||||
console.error(`[federation] ${row.client_id} unreachable:`, err);
|
||||
}
|
||||
}
|
||||
do {
|
||||
queuedMeanwhile.delete(clientId);
|
||||
await sendDue(clientId);
|
||||
} while (queuedMeanwhile.has(clientId));
|
||||
} catch (err) {
|
||||
console.error(`[federation] delivery to ${clientId} failed:`, err);
|
||||
} finally {
|
||||
delivering = false;
|
||||
delivering.delete(clientId);
|
||||
}
|
||||
}
|
||||
|
||||
/** Runs one pass of {@link deliverTo}. */
|
||||
async function sendDue(clientId: string): Promise<void> {
|
||||
for (;;) {
|
||||
const row = db.prepare(
|
||||
`SELECT id, event, attempts, next_attempt_at FROM federation_outbox
|
||||
WHERE client_id = ? ORDER BY created_at, rowid LIMIT 1;`,
|
||||
).get(clientId) as
|
||||
| { id: string; event: string; attempts: number; next_attempt_at: string }
|
||||
| undefined;
|
||||
if (!row || row.next_attempt_at > new Date().toISOString()) return;
|
||||
try {
|
||||
if (!await postEvent(clientId, row.event)) {
|
||||
console.error(`[federation] ${clientId} refused an event`);
|
||||
}
|
||||
db.prepare(`DELETE FROM federation_outbox WHERE id = ?;`).run(row.id);
|
||||
} catch (err) {
|
||||
const wait = Math.min(60_000 * 2 ** row.attempts, MAX_RETRY_MS);
|
||||
db.prepare(
|
||||
`UPDATE federation_outbox SET attempts = attempts + 1, next_attempt_at = ? WHERE id = ?;`,
|
||||
).run(new Date(Date.now() + wait).toISOString(), row.id);
|
||||
console.error(`[federation] ${clientId} unreachable:`, err);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user