Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283import { parseMultikey } from "@atproto/crypto";import { Hono } from "hono";import { bodyLimit } from "hono/body-limit";import type { Context } from "hono";import { decodeJwt, decodeProtectedHeader, importJWK, jwtVerify } from "jose";import { config } from "../config.js";import { appDid } from "./keys.js";import { applyCallbackFact } from "./store.js";
/** * Relay grant-change callback: POST /xrpc/pub.atmo.notify.subscriberChanged. * Inert until atmo.pub registers Airglow's callback URL relay-side; once that * happens, grant changes land here instead of being discovered on failed * sends. * * Verification is pin-then-resolve: the token's `iss` must equal the * CONFIGURED relay DID before any DID resolution happens. Resolving the * token's own `iss` first would let an attacker point us at their key * (key substitution) or at an arbitrary host (SSRF). */
export const SUBSCRIBER_CHANGED_NSID = "pub.atmo.notify.subscriberChanged";export const SUBSCRIBER_CHANGED_PATH = `/xrpc/${SUBSCRIBER_CHANGED_NSID}`;
// The body is one DID, a boolean, and a datetime; a few KB is generous.const MAX_BODY_BYTES = 4096;// Matches the relay's own 5-minute DID cache, so a relay key rotation// propagates within one TTL.const KEY_CACHE_TTL_MS = 5 * 60_000;// Tolerated clock skew for future-dated `iat` and `changedAt` values.const CLOCK_SKEW_MS = 5 * 60_000;// Cap the did.json resolution fetch: an unauthenticated caller must not be// able to pin a hung outbound connection by pointing us at a slow host.const RELAY_FETCH_TIMEOUT_MS = 10_000;// Negative-cache window for a failed resolution. Short, so a transient relay// blip recovers quickly, but long enough that a down or slow relay collapses// into one fetch per window instead of one per incoming request wave.const FAILURE_CACHE_TTL_MS = 10_000;
// `key: null` is a negatively cached failure: a resolution that failed within// the last FAILURE_CACHE_TTL_MS, short-circuiting repeated request-path fetches.let cachedRelayKey: { key: CryptoKey | null; expiresAt: number } | null = null;// Single-flight guard: a cold-cache flood (an attacker can drive resolution by// sending well-formed-but-unsigned tokens, which clear claim checks before the// signature step) must collapse into one outbound did.json fetch, not one per// request. Concurrent callers await the same in-flight resolution.let inFlightRelayKey: Promise<CryptoKey | null> | null = null;// Bumped on every cache clear. A resolution captures the generation it started// under and only writes its result back if the generation still matches, so a// fetch that was in flight when the cache was cleared can't resurrect stale// state over a fresh resolution started after the clear.let relayKeyGeneration = 0;
/** Test hook: drop the cached relay verification key. */export function clearRelayKeyCache(): void { cachedRelayKey = null; inFlightRelayKey = null; relayKeyGeneration++;}
/** * Warm the relay verification key at startup so the first real callback (and an * unauthenticated cold-cache flood) never pays for the outbound did.json fetch * on the request path. Best-effort: resolution failures are swallowed and * retried lazily on first use, and single-flight still bounds any request-path * fetches to one at a time. */export async function prewarmRelayKey(): Promise<void> { await resolveRelayKey(Date.now());}
/** did:web resolution of the PINNED relay DID only, with the cache + single- * flight checks; the actual fetch/decode lives in `fetchRelayKey`. */async function resolveRelayKey(now: number): Promise<CryptoKey | null> { if (cachedRelayKey && cachedRelayKey.expiresAt > now) return cachedRelayKey.key; // Coalesce concurrent cold-cache resolutions into a single fetch, then cache: // a success for KEY_CACHE_TTL_MS, an unreachable/erroring relay for the // shorter FAILURE_CACHE_TTL_MS so an outage can't trigger a fetch per request // wave. The TTL is anchored to fetch completion (a coalesced wave shares one // timestamp). `.finally` clears the guard so the next resolution after the // cache expires can run again. const generation = relayKeyGeneration; // A cache clear bumps the generation and hands the slot to a newer resolution, // so this goes false once a clear races our in-flight fetch. const isCurrent = () => generation === relayKeyGeneration; inFlightRelayKey ??= fetchRelayKey() .then((result) => { // Cache was cleared mid-flight: skip the cache write so we don't clobber a // newer resolution. Callers already coalesced onto this promise still get // the resolved key; only the write-back is dropped. if (!isCurrent()) return result.key; if (result.key) { cachedRelayKey = { key: result.key, expiresAt: Date.now() + KEY_CACHE_TTL_MS }; } else if (result.unreachable) { // Negative-cache ONLY an unreachable or erroring relay: that is the // expensive case, and caching it bounds retry storms during an outage. // A reachable-but-malformed did.json (a 200 with no usable key) is left // uncached, so a valid callback recovers the moment the relay serves a // good document instead of being rejected for a full failure window. cachedRelayKey = { key: null, expiresAt: Date.now() + FAILURE_CACHE_TTL_MS }; } return result.key; }) .finally(() => { // Only clear the slot if it's still ours; a clear after this fetch started // may have installed a newer in-flight resolution we must not null out. if (isCurrent()) inFlightRelayKey = null; }); return inFlightRelayKey;}
// Result of a resolution attempt. `unreachable` distinguishes the relay being// down/erroring (worth negative-caching to bound retries) from a reachable// relay that served an unusable document (recover on the next request).type RelayKeyResult = { key: CryptoKey } | { key: null; unreachable: boolean };
/** Fetch the pinned relay's did.json over HTTPS and decode the #atproto * Multikey (multibase P-256) into a verification key. Returns the key, or a * null result tagged with whether the relay was unreachable. Pure fetch/decode: * caching is owned by `resolveRelayKey`. */async function fetchRelayKey(): Promise<RelayKeyResult> { // Boot validation in lib/config.ts guarantees the relay DID is did:web. const host = decodeURIComponent(config.atmoRelayDid.slice("did:web:".length)); let res: Response; try { res = await fetch(`https://${host}/.well-known/did.json`, { headers: { Accept: "application/json" }, signal: AbortSignal.timeout(RELAY_FETCH_TIMEOUT_MS), }); } catch (err) { // Network error or timeout: the relay is unreachable, so negative-cache to // bound retry storms during an outage. console.warn( `atmo callback: relay did.json unreachable: ${err instanceof Error ? err.message : "error"}`, ); return { key: null, unreachable: true }; } if (!res.ok) { // warn, not debug: a relay being unreachable is an operational event that // must be visible at prod log level (and it drives 401s for up to one // negative-cache window). Bounded to ~one log per window by single-flight. console.warn(`atmo callback: relay did.json fetch failed (${res.status})`); return { key: null, unreachable: true }; }
let doc: { verificationMethod?: Array<{ id?: string; publicKeyMultibase?: string }> }; try { doc = (await res.json()) as typeof doc; } catch (err) { // A reachable relay (2xx) that served a body we can't parse is malformed, // not down: treat it like the no-#atproto-method case below and DON'T // negative-cache it, so a valid callback recovers the moment the relay // serves a good document. A SyntaxError means the body was received but // isn't JSON; an aborted/stalled body read (timeout, reset) is a genuine // outage and stays negative-cached. if (err instanceof SyntaxError) { console.debug("atmo callback: relay did.json body is not valid JSON"); return { key: null, unreachable: false }; } console.warn( `atmo callback: relay did.json read failed: ${err instanceof Error ? err.message : "error"}`, ); return { key: null, unreachable: true }; }
const method = doc.verificationMethod?.find( (m) => typeof m.id === "string" && m.id.endsWith("#atproto"), ); if (!method?.publicKeyMultibase) { console.debug("atmo callback: relay did.json has no #atproto verification method"); return { key: null, unreachable: false }; }
try { const { jwtAlg, keyBytes } = parseMultikey(method.publicKeyMultibase); // keyBytes is the uncompressed SEC1 point: 0x04 || x (32) || y (32). if (jwtAlg !== "ES256" || keyBytes.length !== 65) return { key: null, unreachable: false }; const key = (await importJWK( { kty: "EC", crv: "P-256", x: Buffer.from(keyBytes.subarray(1, 33)).toString("base64url"), y: Buffer.from(keyBytes.subarray(33, 65)).toString("base64url"), }, "ES256", )) as CryptoKey; return { key }; } catch { console.debug("atmo callback: relay key decode failed"); return { key: null, unreachable: false }; }}
/** Validate the bearer token. Claim checks run on the UNVERIFIED decode so * garbage is rejected before any resolution work; the signature is then * verified against the pinned relay's resolved key, which re-validates the * claims. Returns a rejection reason (debug log only, never the response) * or null when the token is good. */async function verifyRelayJwt(header: string | undefined, now: number): Promise<string | null> { if (!header?.startsWith("Bearer ")) return "missing bearer token"; const token = header.slice("Bearer ".length);
let claims: ReturnType<typeof decodeJwt>; let alg: string | undefined; try { claims = decodeJwt(token); alg = decodeProtectedHeader(token).alg; } catch { return "malformed token"; }
if (alg !== "ES256") return "wrong alg"; if (claims.iss !== config.atmoRelayDid) return "issuer is not the pinned relay DID"; if (claims.aud !== appDid()) return "wrong audience"; if (claims.lxm !== SUBSCRIBER_CHANGED_NSID) return "wrong lxm"; if (typeof claims.exp !== "number" || claims.exp * 1000 <= now) return "missing or expired exp"; if (typeof claims.iat === "number" && claims.iat * 1000 > now + CLOCK_SKEW_MS) { return "iat too far in the future"; }
const key = await resolveRelayKey(now); if (!key) return "relay key resolution failed";
try { await jwtVerify(token, key, { algorithms: ["ES256"], issuer: config.atmoRelayDid, audience: appDid(), requiredClaims: ["exp"], }); } catch { return "signature verification failed"; } return null;}
async function handleSubscriberChanged(c: Context): Promise<Response> { const now = Date.now();
const rejection = await verifyRelayJwt(c.req.header("authorization"), now); if (rejection) { console.debug(`atmo subscriberChanged rejected: ${rejection}`); return c.json({ error: "Unauthorized" }, 401); }
let body: Record<string, unknown>; try { body = (await c.req.json()) as Record<string, unknown>; } catch { return c.json({ error: "InvalidRequest" }, 400); } const { recipient, enabled, changedAt } = body ?? {}; if ( typeof recipient !== "string" || !recipient.startsWith("did:") || typeof enabled !== "boolean" || typeof changedAt !== "string" ) { return c.json({ error: "InvalidRequest" }, 400); } const changed = new Date(changedAt); // A far-future changedAt would wedge the row: every later real event would // read as stale. if (Number.isNaN(changed.getTime()) || changed.getTime() > now + CLOCK_SKEW_MS) { return c.json({ error: "InvalidRequest" }, 400); }
// All three outcomes (applied, stale, unknown-user) answer the same way: // deliveries are idempotent and the response leaks nothing about which // DIDs are Airglow users. await applyCallbackFact(recipient, enabled, changed); return c.json({ ok: true });}
/** POST-only sub-app, mounted at SUBSCRIBER_CHANGED_PATH from * lib/runtime/web.ts (which adds the rate limiter). Other verbs fall * through to the host app's 404. */export const subscriberChangedApp = new Hono().post( "/", bodyLimit({ maxSize: MAX_BODY_BYTES }), handleSubscriberChanged,);