feat: implement token source, key store, verifier, snapshot client and event receiver

README.md

@@ -8,17 +8,15 @@ SDK for integrating new projects with PoliNetwork's identity provider (auth.poli
 - [docs/idp-adoption.md](./docs/idp-adoption.md): how the IdP can import the shared contract.
 - [docs/releasing.md](./docs/releasing.md): automated releases and commit conventions.
 
-> **Status: scaffolding.** Snapshot parsing and the actor helpers are implemented; the other modules expose their final API but throw `not_implemented`.
-
 | Export                                         | For                       | What it does                                                                          | Status      |
 | ---------------------------------------------- | ------------------------- | ------------------------------------------------------------------------------------- | ----------- |
 | `idpEndpoints`, contract constants             | everyone                  | IdP URLs and the values fixed in the IdP's code                                       | ✅          |
 | `@polinetwork/auth-kit/contract`               | the IdP                   | Constants, schemas, producer types and event claims only; no network clients         | ✅          |
-| `serviceTokenSource`                           | bot, website, backend     | Client credentials with `private_key_jwt`, cached and renewed at 50% of the lifetime  | stub        |
-| `createKeyStore`                               | resource servers          | JWKS cache that keeps its last good key set during an IdP outage                      | stub        |
-| `verifyAccessToken`, `resolveActor`            | resource servers          | RFC §6.4 token checks, then the request's `Actor` (including `X-PN-Actor`)            | stub        |
-| `accessSnapshot`                               | resource servers          | Local copy of a projection: polling, ETag, single-flight, persistence, `MAX_STALE`    | stub        |
-| `createAccessEventHandler`                     | resource servers          | Fetch API receiver for the IdP's access-changed events                                | stub        |
+| `serviceTokenSource`                           | bot, website, backend     | Client credentials with `private_key_jwt`, cached and renewed at 50% of the lifetime  | ✅          |
+| `createKeyStore`                               | resource servers          | JWKS cache that keeps its last good key set during an IdP outage                      | ✅          |
+| `verifyAccessToken`, `resolveActor`            | resource servers          | RFC §6.4 token checks, then the request's `Actor` (including `X-PN-Actor`)            | ✅          |
+| `accessSnapshot`                               | resource servers          | Local copy of a projection: polling, ETag, single-flight, persistence, `MAX_STALE`    | ✅          |
+| `createAccessEventHandler`                     | resource servers          | Fetch API receiver for the IdP's access-changed events                                | ✅          |
 | `parseAccessSnapshot`, `AccessIndex`           | resource servers          | Validates a snapshot and answers lookups, enforcing `validUntil`                      | ✅          |
 | `formatTelegramActor`, `parseTelegramActor`    | bot, backend              | The `X-PN-Actor: telegram:<id>` header                                                | ✅          |
 | `Persistence`, `memoryPersistence`             | resource servers          | Storage interface for the last good key set and snapshot                              | ✅          |

src/errors.ts

@@ -3,8 +3,8 @@ export type AuthKitErrorCode =
   | "token_invalid"
   | "token_request_failed"
   | "keys_unavailable"
+  | "snapshot_unavailable"
   | "actor_forbidden"
-  | "not_implemented"
 
 /** Every error the SDK throws on purpose. `code` is stable; `message` is for logs only. */
 export class AuthKitError extends Error {
@@ -18,8 +18,3 @@ export class AuthKitError extends Error {
     super(message, options)
   }
 }
-
-/** Placeholder for scaffolded modules; removed as each module is implemented. */
-export function notImplemented(feature: string): never {
-  throw new AuthKitError("not_implemented", `${feature} is not implemented yet`)
-}

src/index.ts

@@ -26,8 +26,17 @@ export {
   DEFAULT_EVENT_PULL_TIMEOUT_MS,
 } from "./snapshot/events"
 export { parseAccessSnapshot } from "./snapshot/parse"
-export { type ServiceTokenSourceOptions, serviceTokenSource, type TokenSource } from "./token-source"
 export {
+  DEFAULT_REFRESH_RATIO,
+  DEFAULT_REQUEST_TIMEOUT_MS,
+  DEFAULT_RETRY_AFTER_FAILURE_MS,
+  type ServiceTokenSourceOptions,
+  serviceTokenSource,
+  type TokenSource,
+} from "./token-source"
+export {
+  DEFAULT_CLOCK_TOLERANCE_SEC,
+  DEFAULT_MAX_TOKEN_AGE,
   type ResolveActorOptions,
   resolveActor,
   type VerifiedToken,

src/key-store.ts

@@ -1,10 +1,13 @@
-import type { JWTVerifyGetKey } from "jose"
-import { notImplemented } from "./errors"
+import { base64url, createLocalJWKSet, errors, type JSONWebKeySet, type JWTVerifyGetKey } from "jose"
+import { z } from "zod"
+import { AuthKitError } from "./errors"
 import type { Persistence } from "./persistence"
 
 export const DEFAULT_KEYS_REFRESH_INTERVAL_MS = 600_000
 export const DEFAULT_KEYS_MAX_AGE_MS = 86_400_000
 export const DEFAULT_UNKNOWN_KID_MIN_INTERVAL_MS = 30_000
+/** Even a token newer than the last fetch cannot trigger fetches closer together than this. */
+const NEW_KEY_MIN_INTERVAL_MS = 1_000
 
 export interface KeyStoreOptions {
   jwksUrl: string
@@ -18,6 +21,8 @@ export interface KeyStoreOptions {
    * last fetch: the IdP signs with a new key as soon as it creates it (docs/contract.md).
    */
   unknownKidMinIntervalMs?: number
+  /** Abandons a JWKS request after this long. Default 10 s. */
+  timeoutMs?: number
   fetch?: typeof fetch
   now?: () => number
   onError?: (error: unknown) => void
@@ -38,7 +43,141 @@ export interface KeyStore {
   status(): KeyStoreStatus
 }
 
+const jwksSchema = z.object({ keys: z.array(z.looseObject({ kty: z.string() })) })
+const persistedSchema = z.object({ fetchedAt: z.number(), jwks: jwksSchema })
+
+type KeySet = {
+  /** When the fetch that produced these keys completed; drives `maxAgeMs`. */
+  fetchedAt: number
+  /** When it started: a key created after this may be missing. */
+  startedAt: number
+  jwks: JSONWebKeySet
+  resolve: ReturnType<typeof createLocalJWKSet>
+}
+
 /** JWKS cache that keeps serving its last good key set during an IdP outage (RFC §6.5). */
-export function createKeyStore(_options: KeyStoreOptions): KeyStore {
-  return notImplemented("createKeyStore")
+export function createKeyStore(options: KeyStoreOptions): KeyStore {
+  const refreshIntervalMs = options.refreshIntervalMs ?? DEFAULT_KEYS_REFRESH_INTERVAL_MS
+  const maxAgeMs = options.maxAgeMs ?? DEFAULT_KEYS_MAX_AGE_MS
+  const unknownKidMinIntervalMs = options.unknownKidMinIntervalMs ?? DEFAULT_UNKNOWN_KID_MIN_INTERVAL_MS
+  const timeoutMs = options.timeoutMs ?? 10_000
+  const doFetch = options.fetch ?? fetch
+  const now = options.now ?? Date.now
+  const persistenceKey = `jwks:${options.jwksUrl}`
+
+  let current: KeySet | null = null
+  let lastAttemptAt: number | null = null
+  let inflight: Promise<void> | null = null
+  let timer: ReturnType<typeof setInterval> | null = null
+
+  function install(jwks: JSONWebKeySet, fetchedAt: number, startedAt: number) {
+    current = { fetchedAt, startedAt, jwks, resolve: createLocalJWKSet(jwks) }
+  }
+
+  async function fetchKeys(): Promise<void> {
+    const startedAt = now()
+    lastAttemptAt = startedAt
+    let jwks: JSONWebKeySet
+    try {
+      const response = await doFetch(options.jwksUrl, {
+        headers: { Accept: "application/json" },
+        signal: AbortSignal.timeout(timeoutMs),
+      })
+      if (!response.ok) {
+        await response.body?.cancel()
+        throw new Error(`JWKS endpoint answered ${response.status}`)
+      }
+      jwks = jwksSchema.parse(await response.json()) as JSONWebKeySet
+    } catch (error) {
+      throw new AuthKitError("keys_unavailable", "Could not fetch the IdP's signing keys", { cause: error })
+    }
+    const fetchedAt = now()
+    install(jwks, fetchedAt, startedAt)
+    try {
+      await options.persistence?.save(persistenceKey, JSON.stringify({ fetchedAt, jwks }))
+    } catch (error) {
+      options.onError?.(error)
+    }
+  }
+
+  /** One fetch at a time; concurrent callers share it. */
+  function refresh(): Promise<void> {
+    inflight ??= fetchKeys().finally(() => {
+      inflight = null
+    })
+    return inflight
+  }
+
+  /** Refetches unless one ran less than `minIntervalMs` ago; resolves either way. */
+  async function refreshAtMostEvery(minIntervalMs: number) {
+    if (inflight) return inflight.catch(() => {})
+    if (lastAttemptAt !== null && now() - lastAttemptAt < minIntervalMs) return
+    await refresh().catch((error) => options.onError?.(error))
+  }
+
+  function fresh(keys: KeySet | null): keys is KeySet {
+    return keys !== null && now() - keys.fetchedAt <= maxAgeMs
+  }
+
+  function issuedAt(token: { payload?: unknown }): number | null {
+    try {
+      if (typeof token.payload !== "string") return null
+      const { iat } = JSON.parse(new TextDecoder().decode(base64url.decode(token.payload))) as { iat?: unknown }
+      return typeof iat === "number" ? iat * 1000 : null
+    } catch {
+      return null
+    }
+  }
+
+  const getKey: JWTVerifyGetKey = async (header, token) => {
+    if (!fresh(current)) await refreshAtMostEvery(unknownKidMinIntervalMs)
+    if (!fresh(current)) throw new AuthKitError("keys_unavailable", "No signing keys fetched recently enough")
+    try {
+      return await current.resolve(header, token)
+    } catch (error) {
+      if (!(error instanceof errors.JWKSNoMatchingKey)) throw error
+      // A token issued since the last fetch started may be signed with a key created since then.
+      // `iat` has one-second resolution, so compare against the start of that second.
+      const iat = issuedAt(token)
+      const lastFetchSecond = Math.floor(current.startedAt / 1000) * 1000
+      const minInterval = iat !== null && iat >= lastFetchSecond ? NEW_KEY_MIN_INTERVAL_MS : unknownKidMinIntervalMs
+      await refreshAtMostEvery(minInterval)
+      if (!fresh(current)) throw new AuthKitError("keys_unavailable", "No signing keys fetched recently enough")
+      return current.resolve(header, token)
+    }
+  }
+
+  return {
+    async start() {
+      try {
+        const stored = await options.persistence?.load(persistenceKey)
+        if (stored) {
+          const parsed = persistedSchema.parse(JSON.parse(stored))
+          // A timestamp from the future would keep old keys valid indefinitely.
+          if (parsed.fetchedAt > now()) throw new Error("Persisted key set is dated in the future")
+          install(parsed.jwks as JSONWebKeySet, parsed.fetchedAt, parsed.fetchedAt)
+        }
+      } catch (error) {
+        options.onError?.(error)
+      }
+      await refresh().catch((error) => options.onError?.(error))
+      if (!timer) {
+        timer = setInterval(() => {
+          refresh().catch((error) => options.onError?.(error))
+        }, refreshIntervalMs)
+        timer.unref?.()
+      }
+    },
+    stop() {
+      if (timer) clearInterval(timer)
+      timer = null
+    },
+    getKey,
+    status() {
+      return {
+        fetchedAt: current?.fetchedAt ?? null,
+        kids: (current?.jwks.keys ?? []).map((key) => key.kid).filter((kid): kid is string => !!kid),
+      }
+    },
+  }
 }

src/snapshot/client.ts

@@ -1,8 +1,10 @@
+import { z } from "zod"
 import type { Actor } from "../actor"
-import { notImplemented } from "../errors"
+import { AuthKitError } from "../errors"
 import type { Persistence } from "../persistence"
 import type { TokenSource } from "../token-source"
 import type { AccessIndex, AccessSubject } from "./access-index"
+import { parseAccessSnapshot } from "./parse"
 
 export const DEFAULT_POLL_INTERVAL_MS = 30_000
 export const DEFAULT_MAX_STALE_MS = 3_600_000
@@ -18,6 +20,8 @@ export interface AccessSnapshotOptions {
   pollIntervalMs?: number
   /** Past this time since the last `200`/`304`, every permission check denies. Default 1 h. */
   maxStaleMs?: number
+  /** Abandons a pull, including reading its body, after this long. Default 10 s. */
+  timeoutMs?: number
   fetch?: typeof fetch
   now?: () => number
   onError?: (error: unknown) => void
@@ -50,10 +54,146 @@ export interface AccessSnapshotClient {
   status(): SnapshotStatus
 }
 
+const persistedSchema = z.object({ body: z.string(), etag: z.string().nullable(), lastSyncAt: z.number() })
+
 /**
  * Keeps a local, always-current copy of a projection (RFC §5): polling with ETag, single-flight
  * refresh, validation before an atomic swap, persistence and the `MAX_STALE` rule.
  */
-export function accessSnapshot(_options: AccessSnapshotOptions): AccessSnapshotClient {
-  return notImplemented("accessSnapshot")
+export function accessSnapshot(options: AccessSnapshotOptions): AccessSnapshotClient {
+  const pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS
+  const maxStaleMs = options.maxStaleMs ?? DEFAULT_MAX_STALE_MS
+  const timeoutMs = options.timeoutMs ?? 10_000
+  const doFetch = options.fetch ?? fetch
+  const now = options.now ?? Date.now
+  const persistenceKey = `snapshot:${options.projection}`
+
+  // Replaced together, so readers never see an index with another snapshot's ETag or body.
+  let state: { index: AccessIndex; body: string; etag: string | null } | null = null
+  let lastSyncAt: number | null = null
+  let running: Promise<void> | null = null
+  let queued: Promise<void> | null = null
+  let timer: ReturnType<typeof setInterval> | null = null
+
+  async function persist() {
+    if (!options.persistence || !state || lastSyncAt === null) return
+    try {
+      await options.persistence.save(persistenceKey, JSON.stringify({ body: state.body, etag: state.etag, lastSyncAt }))
+    } catch (error) {
+      options.onError?.(error)
+    }
+  }
+
+  async function request(token: string): Promise<Response> {
+    const headers: Record<string, string> = { Authorization: `Bearer ${token}`, Accept: "application/json" }
+    // Only an ETag the IdP returned: `If-None-Match: *` would always answer 304.
+    if (state?.etag) headers["If-None-Match"] = state.etag
+    try {
+      return await doFetch(options.url, { headers, signal: AbortSignal.timeout(timeoutMs) })
+    } catch (error) {
+      throw new AuthKitError("snapshot_unavailable", "Snapshot endpoint unreachable", { cause: error })
+    }
+  }
+
+  async function pull(): Promise<void> {
+    let response = await request(await options.tokens.getToken())
+    if (response.status === 401) {
+      await response.body?.cancel()
+      options.tokens.invalidate()
+      response = await request(await options.tokens.getToken())
+    }
+    if (response.status === 304) {
+      if (!state?.etag) {
+        throw new AuthKitError("snapshot_unavailable", "Snapshot endpoint answered 304 without a snapshot and ETag")
+      }
+      lastSyncAt = now()
+      await persist()
+      return
+    }
+    if (response.status !== 200) {
+      await response.body?.cancel()
+      throw new AuthKitError("snapshot_unavailable", `Snapshot endpoint answered ${response.status}`)
+    }
+    let body: string
+    try {
+      body = await response.text()
+    } catch (error) {
+      throw new AuthKitError("snapshot_unavailable", "Snapshot body could not be read", { cause: error })
+    }
+    // Throws snapshot_invalid; the previous snapshot stays in place.
+    const index = parseAccessSnapshot(body, options.projection)
+    state = { index, body, etag: response.headers.get("etag") }
+    lastSyncAt = now()
+    await persist()
+  }
+
+  function refresh(): Promise<void> {
+    if (!running) {
+      running = pull().finally(() => {
+        running = null
+      })
+      return running
+    }
+    // A trigger during a pull may announce a change the running pull has not seen.
+    queued ??= running
+      .catch(() => {})
+      .then(() => {
+        queued = null
+        return refresh()
+      })
+    return queued
+  }
+
+  function fresh() {
+    return lastSyncAt !== null && now() - lastSyncAt <= maxStaleMs
+  }
+
+  function decisionIndex(): AccessIndex | null {
+    return state && fresh() ? state.index : null
+  }
+
+  return {
+    async start() {
+      try {
+        const stored = await options.persistence?.load(persistenceKey)
+        if (stored) {
+          const parsed = persistedSchema.parse(JSON.parse(stored))
+          // A timestamp from the future would keep an old snapshot fresh indefinitely.
+          if (parsed.lastSyncAt > now()) throw new Error("Persisted snapshot is dated in the future")
+          state = { index: parseAccessSnapshot(parsed.body, options.projection), body: parsed.body, etag: parsed.etag }
+          lastSyncAt = parsed.lastSyncAt
+        }
+      } catch (error) {
+        options.onError?.(error)
+      }
+      await refresh().catch((error) => options.onError?.(error))
+      if (!timer) {
+        timer = setInterval(() => {
+          refresh().catch((error) => options.onError?.(error))
+        }, pollIntervalMs)
+        timer.unref?.()
+      }
+    },
+    stop() {
+      if (timer) clearInterval(timer)
+      timer = null
+    },
+    refresh,
+    current: () => state?.index ?? null,
+    subjectBySub: (sub) => decisionIndex()?.bySub(sub),
+    subjectByTelegramId: (telegramId) => decisionIndex()?.byTelegramId(telegramId),
+    has(actor, permission) {
+      if (actor.kind === "service" || actor.sub === null) return false
+      const index = decisionIndex()
+      return index ? index.has(index.bySub(actor.sub), permission, now()) : false
+    },
+    status() {
+      return {
+        lastSyncAt,
+        fresh: fresh(),
+        generation: state?.index.generation ?? null,
+        subjects: state?.index.size ?? 0,
+      }
+    },
+  }
 }