From 2c3d6b1716ffc05591356523d031496744f9724a Mon Sep 17 00:00:00 2001 From: Bianca Date: Fri, 9 Oct 2026 21:11:36 +0200 Subject: [PATCH 1/3] feat: implement token source, key store, verifier, snapshot client and event receiver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the scaffolded stubs with the RFC v3 §5, §6 and §14.1 implementations: - serviceTokenSource: client credentials with private_key_jwt, cached, renewed in the background at half the lifetime, with failure backoff and timeouts. - createKeyStore: last-known-good JWKS with persistence, a 24 h maximum age and the contract's unknown-kid refetch rules. - verifyAccessToken / resolveActor: the §6.4 checks and X-PN-Actor handling. - accessSnapshot: ETag polling, single-flight refresh with one queued follow-up, validation before swap, persistence and MAX_STALE. - createAccessEventHandler: SET receiver with a bounded body. The not_implemented error code, used only by the stubs, is removed. --- README.md | 12 +- src/errors.ts | 7 +- src/index.ts | 11 +- src/key-store.ts | 147 ++++++++++++++++++++++++- src/snapshot/client.ts | 144 +++++++++++++++++++++++- src/snapshot/events.ts | 87 ++++++++++++++- src/token-source.ts | 138 ++++++++++++++++++++++- src/verify.ts | 67 +++++++++-- tests/events.test.ts | 106 ++++++++++++++++++ tests/idp.ts | 97 ++++++++++++++++ tests/key-store.test.ts | 127 +++++++++++++++++++++ tests/snapshot-client.test.ts | 202 ++++++++++++++++++++++++++++++++++ tests/token-source.test.ts | 149 +++++++++++++++++++++++++ tests/verify.test.ts | 128 +++++++++++++++++++++ 14 files changed, 1385 insertions(+), 37 deletions(-) create mode 100644 tests/events.test.ts create mode 100644 tests/idp.ts create mode 100644 tests/key-store.test.ts create mode 100644 tests/snapshot-client.test.ts create mode 100644 tests/token-source.test.ts create mode 100644 tests/verify.test.ts diff --git a/README.md b/README.md index 2ab620b..76862ff 100644 --- a/README.md +++ b/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:` header | ✅ | | `Persistence`, `memoryPersistence` | resource servers | Storage interface for the last good key set and snapshot | ✅ | diff --git a/src/errors.ts b/src/errors.ts index eab7ce6..07f7836 100644 --- a/src/errors.ts +++ b/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`) -} diff --git a/src/index.ts b/src/index.ts index c9f74b5..8e0af89 100644 --- a/src/index.ts +++ b/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, diff --git a/src/key-store.ts b/src/key-store.ts index 3984102..c763701 100644 --- a/src/key-store.ts +++ b/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 +} + /** 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 | null = null + let timer: ReturnType | null = null + + function install(jwks: JSONWebKeySet, fetchedAt: number, startedAt: number) { + current = { fetchedAt, startedAt, jwks, resolve: createLocalJWKSet(jwks) } + } + + async function fetchKeys(): Promise { + 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 { + 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), + } + }, + } } diff --git a/src/snapshot/client.ts b/src/snapshot/client.ts index 1d5c81d..d9a1408 100644 --- a/src/snapshot/client.ts +++ b/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,144 @@ 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 | null = null + let queued: Promise | null = null + let timer: ReturnType | 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 { + const headers: Record = { 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 { + 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) throw new AuthKitError("snapshot_unavailable", "Snapshot endpoint answered 304 without a snapshot") + 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 { + 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, + } + }, + } } diff --git a/src/snapshot/events.ts b/src/snapshot/events.ts index f9fd45a..98a3f07 100644 --- a/src/snapshot/events.ts +++ b/src/snapshot/events.ts @@ -1,9 +1,14 @@ -import { notImplemented } from "../errors" +import { type JWTPayload, jwtVerify } from "jose" +import { ACCESS_CHANGED_EVENT, EVENT_CONTENT_TYPE, SIGNING_ALGORITHMS, TOKEN_TYPES } from "../contract/constants" +import { accessChangedEventSchema } from "../contract/event" +import { AuthKitError } from "../errors" import type { KeyStore } from "../key-store" import type { AccessSnapshotClient } from "./client" /** Stays below the IdP's 8 s delivery timeout (`IDP_EVENT_TIMEOUT_MS`). */ export const DEFAULT_EVENT_PULL_TIMEOUT_MS = 6_000 +/** An event is a signed "pull now" of a few hundred bytes. */ +const MAX_EVENT_BYTES = 16_384 export interface AccessEventHandlerOptions { keyStore: Pick @@ -14,6 +19,35 @@ export interface AccessEventHandlerOptions { snapshot: Pick /** How long to wait for the triggered pull before answering `503`. */ pullTimeoutMs?: number + clockToleranceSec?: number + /** Reports rejected events and failed pulls. */ + onError?: (error: unknown) => void +} + +/** The body as text, or null once it exceeds `limit` bytes; stops reading at the limit. */ +async function readAtMost(request: Request, limit: number): Promise { + if (Number(request.headers.get("content-length") ?? "0") > limit) return null + if (!request.body) return "" + const reader = request.body.getReader() + const decoder = new TextDecoder() + let text = "" + let size = 0 + for (;;) { + const { done, value } = await reader.read() + if (done) return text + decoder.decode() + size += value.byteLength + if (size > limit) { + await reader.cancel() + return null + } + text += decoder.decode(value, { stream: true }) + } +} + +function reply(status: number, err?: string, description?: string): Response { + if (!err) return new Response(null, { status }) + // RFC 8935 §2.3 error body. + return Response.json({ err, description }, { status }) } /** @@ -25,6 +59,53 @@ export interface AccessEventHandlerOptions { * keeps running), `401` for an invalid token. Never deduplicates by `jti`: the IdP reuses it on * retries that may cover newer changes. */ -export function createAccessEventHandler(_options: AccessEventHandlerOptions): (request: Request) => Promise { - return notImplemented("createAccessEventHandler") +export function createAccessEventHandler(options: AccessEventHandlerOptions): (request: Request) => Promise { + const pullTimeoutMs = options.pullTimeoutMs ?? DEFAULT_EVENT_PULL_TIMEOUT_MS + + return async (request) => { + if (request.method !== "POST") return new Response(null, { status: 405, headers: { Allow: "POST" } }) + const contentType = request.headers.get("content-type")?.split(";")[0]?.trim().toLowerCase() + if (contentType !== EVENT_CONTENT_TYPE) return reply(415, "invalid_request", "Expected a security event token") + const raw = await readAtMost(request, MAX_EVENT_BYTES) + if (raw === null) return reply(413, "invalid_request", "Event too large") + const body = raw.trim() + if (body.length === 0) return reply(400, "invalid_request", "Invalid body") + + let claims: JWTPayload + try { + ;({ payload: claims } = await jwtVerify(body, options.keyStore.getKey, { + issuer: options.issuer, + audience: options.audience, + typ: TOKEN_TYPES.event, + algorithms: [...SIGNING_ALGORITHMS], + requiredClaims: ["iat", "exp", "jti"], + clockTolerance: options.clockToleranceSec ?? 30, + })) + } catch (error) { + options.onError?.(error) + // Without keys the event may be valid: let the IdP retry instead of dropping it. + if (error instanceof AuthKitError && error.code === "keys_unavailable") return reply(503) + return reply(401, "authentication_failed", "Invalid security event token") + } + const event = accessChangedEventSchema.safeParse(claims) + if (!event.success || event.data.events[ACCESS_CHANGED_EVENT].projection !== options.projection) + return reply(400, "invalid_request", "Unexpected event") + + let timeout: ReturnType | undefined + const pulled = options.snapshot.refresh().then( + () => true, + (error) => { + options.onError?.(error) + return false + } + ) + const timedOut = new Promise((resolve) => { + timeout = setTimeout(() => resolve(false), pullTimeoutMs) + }) + try { + return (await Promise.race([pulled, timedOut])) ? reply(202) : reply(503) + } finally { + clearTimeout(timeout) + } + } } diff --git a/src/token-source.ts b/src/token-source.ts index 3451b7b..e455ec3 100644 --- a/src/token-source.ts +++ b/src/token-source.ts @@ -1,5 +1,6 @@ -import type { JWK } from "jose" -import { notImplemented } from "./errors" +import { importJWK, type JWK, SignJWT } from "jose" +import { CLIENT_ASSERTION_TYPE } from "./contract/constants" +import { AuthKitError } from "./errors" export interface TokenSource { /** A valid access token, from cache or freshly issued. Concurrent callers share one request. */ @@ -24,11 +25,140 @@ export interface ServiceTokenSourceOptions { scopes: readonly string[] /** Fraction of the token lifetime after which it is renewed. Default 0.5. */ refreshRatio?: number + /** After a failed request, callers get the cached token or the error until this has passed. Default 5 s. */ + retryAfterFailureMs?: number + /** Abandons a token request after this long. Default 10 s. */ + timeoutMs?: number fetch?: typeof fetch now?: () => number + /** Reports a failed renewal while a cached token is still being served. */ + onError?: (error: unknown) => void } +export const DEFAULT_REFRESH_RATIO = 0.5 +export const DEFAULT_RETRY_AFTER_FAILURE_MS = 5_000 +export const DEFAULT_REQUEST_TIMEOUT_MS = 10_000 +/** Client assertions are single-use and short-lived. */ +const ASSERTION_LIFETIME_SEC = 60 +/** A cached token is not handed out in its last seconds, so it cannot expire in flight. */ +const EXPIRY_MARGIN_MS = 10_000 +const SIGNING_ALGORITHMS = new Set(["EdDSA", "ES256"]) + +type CachedToken = { value: string; refreshAt: number; expiresAt: number } + /** Client credentials with `private_key_jwt` (RFC 7523), cached and renewed early (RFC §14.1). */ -export function serviceTokenSource(_options: ServiceTokenSourceOptions): TokenSource { - return notImplemented("serviceTokenSource") +export function serviceTokenSource(options: ServiceTokenSourceOptions): TokenSource { + const { clientId, privateKey } = options + const alg = privateKey.alg + if (!alg || !SIGNING_ALGORITHMS.has(alg)) throw new TypeError("privateKey.alg must be EdDSA or ES256") + if (!privateKey.kid) throw new TypeError("privateKey.kid is required") + if (!privateKey.d) throw new TypeError("privateKey must be a private key") + const kid = privateKey.kid + const refreshRatio = options.refreshRatio ?? DEFAULT_REFRESH_RATIO + if (!(refreshRatio > 0 && refreshRatio <= 1)) throw new TypeError("refreshRatio must be in (0, 1]") + const retryAfterFailureMs = options.retryAfterFailureMs ?? DEFAULT_RETRY_AFTER_FAILURE_MS + const timeoutMs = options.timeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS + const doFetch = options.fetch ?? fetch + const now = options.now ?? Date.now + const signingKey = importJWK(privateKey, alg) + // Surfaced by the first request instead of as an unhandled rejection. + signingKey.catch(() => {}) + + let cached: CachedToken | null = null + let inflight: Promise | null = null + let lastFailure: { at: number; error: unknown } | null = null + + async function assertion(issuedAt: number): Promise { + return new SignJWT({}) + .setProtectedHeader({ alg: alg as string, kid }) + .setIssuer(clientId) + .setSubject(clientId) + .setAudience(options.tokenAudience) + .setIssuedAt(issuedAt) + .setExpirationTime(issuedAt + ASSERTION_LIFETIME_SEC) + .setJti(crypto.randomUUID()) + .sign(await signingKey) + } + + async function request(): Promise { + const startedAt = now() + const body = new URLSearchParams({ + grant_type: "client_credentials", + client_id: clientId, + client_assertion_type: CLIENT_ASSERTION_TYPE, + client_assertion: await assertion(Math.floor(startedAt / 1000)), + resource: options.resource, + scope: options.scopes.join(" "), + }) + let response: Response + try { + response = await doFetch(options.tokenUrl, { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded", Accept: "application/json" }, + body, + signal: AbortSignal.timeout(timeoutMs), + }) + } catch (error) { + throw new AuthKitError("token_request_failed", "Token endpoint unreachable", { cause: error }) + } + const json = (await response.json().catch(() => null)) as Record | null + if (!response.ok) { + // Only the OAuth error code: the description may echo request details. + const code = typeof json?.error === "string" ? json.error : "unknown_error" + throw new AuthKitError("token_request_failed", `Token endpoint answered ${response.status} (${code})`) + } + const token = json?.access_token + const expiresIn = json?.expires_in + const tokenType = json?.token_type + if ( + typeof token !== "string" || + token.length === 0 || + typeof expiresIn !== "number" || + !(expiresIn > 0) || + (typeof tokenType === "string" && tokenType.toLowerCase() !== "bearer") + ) + throw new AuthKitError("token_request_failed", "Token endpoint returned an unusable response") + const lifetimeMs = expiresIn * 1000 + cached = { value: token, refreshAt: startedAt + lifetimeMs * refreshRatio, expiresAt: startedAt + lifetimeMs } + lastFailure = null + return token + } + + function usable(token: CachedToken | null, at: number): token is CachedToken { + return token !== null && at < token.expiresAt - EXPIRY_MARGIN_MS + } + + /** One request at a time; a failure starts the backoff window. */ + function renew(): Promise { + inflight ??= request() + .catch((error: unknown) => { + lastFailure = { at: now(), error } + throw error + }) + .finally(() => { + inflight = null + }) + return inflight + } + + function backingOff(at: number) { + return lastFailure !== null && at - lastFailure.at < retryAfterFailureMs + } + + return { + async getToken() { + const at = now() + if (usable(cached, at)) { + // Past the renewal point a still-valid token is handed out at once while a new one is fetched. + if (at >= cached.refreshAt && !inflight && !backingOff(at)) renew().catch((error) => options.onError?.(error)) + return cached.value + } + if (backingOff(at)) throw lastFailure?.error + return renew() + }, + invalidate() { + cached = null + lastFailure = null + }, + } } diff --git a/src/verify.ts b/src/verify.ts index 7aa79a2..ef513bb 100644 --- a/src/verify.ts +++ b/src/verify.ts @@ -1,6 +1,7 @@ -import type { JWTPayload } from "jose" -import type { Actor } from "./actor" -import { notImplemented } from "./errors" +import { type JWTPayload, jwtVerify } from "jose" +import { type Actor, parseTelegramActor } from "./actor" +import { SIGNING_ALGORITHMS, SUBJECT_TYPE_CLAIM, TOKEN_TYPES } from "./contract/constants" +import { AuthKitError } from "./errors" import type { KeyStore } from "./key-store" import type { AccessIndex } from "./snapshot/access-index" @@ -22,18 +23,47 @@ export type VerifiedToken = { claims: JWTPayload } +export const DEFAULT_CLOCK_TOLERANCE_SEC = 30 +export const DEFAULT_MAX_TOKEN_AGE = "2h" +const REQUIRED_CLAIMS = ["exp", "iat", "sub", "client_id", "scope", "jti", SUBJECT_TYPE_CLAIM] + /** * The RFC §6.4 checks: issuer, audience, `typ: at+jwt`, EdDSA only, required claims, clock * tolerance, and `pn_subject_type` consistent with `sub` and `client_id`. - * Throws `token_invalid`. + * Throws `token_invalid`, or `keys_unavailable` when no usable key set is available. */ -export async function verifyAccessToken(_token: string, _options: VerifyAccessTokenOptions): Promise { - return notImplemented("verifyAccessToken") +export async function verifyAccessToken(token: string, options: VerifyAccessTokenOptions): Promise { + let claims: JWTPayload + try { + ;({ payload: claims } = await jwtVerify(token, options.keyStore.getKey, { + issuer: options.issuer, + audience: options.audience, + typ: TOKEN_TYPES.access, + algorithms: [...SIGNING_ALGORITHMS], + requiredClaims: REQUIRED_CLAIMS, + clockTolerance: options.clockToleranceSec ?? DEFAULT_CLOCK_TOLERANCE_SEC, + maxTokenAge: options.maxTokenAge ?? DEFAULT_MAX_TOKEN_AGE, + })) + } catch (error) { + if (error instanceof AuthKitError) throw error + throw new AuthKitError("token_invalid", "Access token failed verification", { cause: error }) + } + + const { sub, client_id: clientId, scope } = claims + const subjectType = claims[SUBJECT_TYPE_CLAIM] + if (typeof sub !== "string" || typeof clientId !== "string" || typeof scope !== "string" || !sub || !clientId) + throw new AuthKitError("token_invalid", "Access token has malformed claims") + let kind: VerifiedToken["kind"] + if (subjectType === "client" && sub === clientId) kind = "service" + else if (subjectType === "user" && sub !== clientId) kind = "user" + else throw new AuthKitError("token_invalid", "Access token subject type does not match its subject") + + return { kind, clientId, sub, scopes: new Set(scope.split(" ").filter(Boolean)), claims } } export interface ResolveActorOptions { /** The raw `X-PN-Actor` header, if present. */ - actorHeader?: string | null + actorHeader?: string | null | undefined /** Scope that allows asserting a Telegram actor, e.g. `backend:tg:act-as`. */ actAsScope: string /** Current snapshot, to fill in `sub` / `telegramId`; null before the first sync. */ @@ -44,6 +74,25 @@ export interface ResolveActorOptions { * Turns a verified token into the request's actor. Only a service token holding `actAsScope` * may assert a Telegram actor; anything else with the header throws `actor_forbidden`. */ -export function resolveActor(_token: VerifiedToken, _options: ResolveActorOptions): Actor { - return notImplemented("resolveActor") +export function resolveActor(token: VerifiedToken, options: ResolveActorOptions): Actor { + const { actorHeader, snapshot } = options + if (actorHeader === undefined || actorHeader === null) { + if (token.kind === "service") return { kind: "service", client: token.clientId } + return { + kind: "user", + client: token.clientId, + sub: token.sub, + telegramId: snapshot?.bySub(token.sub)?.telegramId ?? null, + } + } + if (token.kind !== "service" || !token.scopes.has(options.actAsScope)) + throw new AuthKitError("actor_forbidden", "This token may not assert an actor") + const telegramId = parseTelegramActor(actorHeader) + if (!telegramId) throw new AuthKitError("actor_forbidden", "Malformed actor header") + return { + kind: "telegram", + client: token.clientId, + telegramId, + sub: snapshot?.byTelegramId(telegramId)?.sub ?? null, + } } diff --git a/tests/events.test.ts b/tests/events.test.ts new file mode 100644 index 0000000..ce08181 --- /dev/null +++ b/tests/events.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it } from "vitest" +import { ACCESS_CHANGED_EVENT, accessChangedEventClaims, createAccessEventHandler, createKeyStore } from "../src" +import { BACKEND, fakeFetch, ISSUER, JWKS_URL, sign, signingKey } from "./idp" + +const key = await signingKey("k1") +const store = createKeyStore({ + jwksUrl: JWKS_URL, + fetch: fakeFetch(() => Response.json({ keys: [key.publicJwk] })).fetch, +}) +await store.start() +store.stop() + +function handler(refresh: () => Promise, pullTimeoutMs = 200) { + let pulls = 0 + const handle = createAccessEventHandler({ + keyStore: store, + issuer: ISSUER, + audience: BACKEND, + projection: "backend", + snapshot: { + refresh: () => { + pulls++ + return refresh() + }, + }, + pullTimeoutMs, + onError: () => {}, + }) + return { handle, pulls: () => pulls } +} + +const event = (overrides: Record = {}, typ = "secevent+jwt") => + sign( + key, + { + ...accessChangedEventClaims({ issuer: ISSUER, audience: BACKEND, jti: "access-1", projection: "backend" }), + ...overrides, + }, + { typ } + ) + +const post = (body: string, contentType = "application/secevent+jwt") => + new Request("http://backend/internal/events", { method: "POST", headers: { "Content-Type": contentType }, body }) + +describe("createAccessEventHandler", () => { + it("pulls and answers 202 for a valid event, every time it arrives", async () => { + const { handle, pulls } = handler(async () => {}) + const token = await event() + expect((await handle(post(token))).status).toBe(202) + expect((await handle(post(token))).status).toBe(202) + expect(pulls()).toBe(2) + }) + + it.each([ + ["an access token", {}, "at+jwt"], + ["another audience", { aud: "https://auth.example.org/api/internal" }, undefined], + ["another issuer", { iss: "https://evil.example/api/auth" }, undefined], + ["an expired event", { iat: 1, exp: 61 }, undefined], + ])("rejects %s with 401 without pulling", async (_name, overrides, typ) => { + const { handle, pulls } = handler(async () => {}) + expect((await handle(post(await event(overrides, typ)))).status).toBe(401) + expect(pulls()).toBe(0) + }) + + it("rejects another projection and other event types", async () => { + const { handle, pulls } = handler(async () => {}) + const other = await event({ events: { [ACCESS_CHANGED_EVENT]: { projection: "rooms" } } }) + expect((await handle(post(other))).status).toBe(400) + const unknown = await event({ events: { "https://example.org/other": {} } }) + expect((await handle(post(unknown))).status).toBe(400) + expect(pulls()).toBe(0) + }) + + it("rejects other methods and content types", async () => { + const { handle } = handler(async () => {}) + expect((await handle(new Request("http://backend/internal/events"))).status).toBe(405) + expect((await handle(post(await event(), "application/json"))).status).toBe(415) + }) + + it("refuses oversized bodies without pulling, even without a Content-Length", async () => { + const { handle, pulls } = handler(async () => {}) + const chunk = new TextEncoder().encode("a".repeat(8_192)) + const stream = new ReadableStream({ + pull(controller) { + controller.enqueue(chunk) + }, + }) + const request = new Request("http://backend/internal/events", { + method: "POST", + headers: { "Content-Type": "application/secevent+jwt" }, + body: stream, + duplex: "half", + } as RequestInit) + expect((await handle(request)).status).toBe(413) + expect(pulls()).toBe(0) + }) + + it("answers 503 when the pull fails or outlasts the budget", async () => { + const failing = handler(async () => { + throw new Error("IdP down") + }) + expect((await failing.handle(post(await event()))).status).toBe(503) + const slow = handler(() => new Promise((resolve) => setTimeout(resolve, 1_000)), 20) + expect((await slow.handle(post(await event()))).status).toBe(503) + }) +}) diff --git a/tests/idp.ts b/tests/idp.ts new file mode 100644 index 0000000..5a54f14 --- /dev/null +++ b/tests/idp.ts @@ -0,0 +1,97 @@ +import { exportJWK, generateKeyPair, type JWK, type JWTPayload, SignJWT } from "jose" + +export const ISSUER = "https://auth.example.org/api/auth" +export const BACKEND = "https://backend.internal.example.org" +export const JWKS_URL = `${ISSUER}/jwks` + +type PrivateKey = Awaited>["privateKey"] + +export type SigningKey = { kid: string; privateKey: PrivateKey; publicJwk: JWK; privateJwk: JWK } + +export async function signingKey(kid: string, alg: "EdDSA" | "ES256" = "EdDSA"): Promise { + const { privateKey, publicKey } = await generateKeyPair(alg, { extractable: true }) + return { + kid, + privateKey, + publicJwk: { ...(await exportJWK(publicKey)), kid, alg, use: "sig" }, + privateJwk: { ...(await exportJWK(privateKey)), kid, alg }, + } +} + +export function sign( + key: SigningKey, + claims: JWTPayload, + header: { typ?: string; alg?: string; kid?: string } = {} +): Promise { + return new SignJWT(claims) + .setProtectedHeader({ + alg: header.alg ?? key.publicJwk.alg ?? "EdDSA", + kid: header.kid ?? key.kid, + typ: header.typ ?? "at+jwt", + }) + .sign(key.privateKey) +} + +export const nowSec = () => Math.floor(Date.now() / 1000) + +export function serviceClaims(overrides: JWTPayload = {}): JWTPayload { + const iat = nowSec() + return { + iss: ISSUER, + aud: BACKEND, + sub: "telegram-bot", + client_id: "telegram-bot", + azp: "telegram-bot", + scope: "backend:tg:read backend:tg:act-as", + pn_subject_type: "client", + jti: crypto.randomUUID(), + iat, + exp: iat + 3600, + ...overrides, + } +} + +/** Claims without the given keys, to test missing claims. */ +export function without(claims: JWTPayload, ...keys: string[]): JWTPayload { + return Object.fromEntries(Object.entries(claims).filter(([key]) => !keys.includes(key))) +} + +export function userClaims(overrides: JWTPayload = {}): JWTPayload { + return serviceClaims({ + sub: "usr_moderator", + client_id: "admin-dashboard", + azp: "admin-dashboard", + aud: [BACKEND, `${ISSUER}/oauth2/userinfo`], + scope: "openid profile email offline_access backend:admin", + pn_subject_type: "user", + ...overrides, + }) +} + +type Handler = (request: Request) => Response | Promise + +/** A `fetch` that records requests and answers from a handler. */ +export function fakeFetch(handler: Handler) { + const requests: Request[] = [] + const fetch = (async (input: string | URL | Request, init?: RequestInit) => { + const request = new Request(input, init) + requests.push(request.clone()) + return handler(request) + }) as typeof globalThis.fetch + return { fetch, requests } +} + +export function jwksHandler(keys: () => SigningKey[]): Handler { + return () => Response.json({ keys: keys().map((key) => key.publicJwk) }) +} + +/** A controllable clock. */ +export function clock(start = Date.now()) { + let at = start + return { + now: () => at, + advance(ms: number) { + at += ms + }, + } +} diff --git a/tests/key-store.test.ts b/tests/key-store.test.ts new file mode 100644 index 0000000..dbf0d18 --- /dev/null +++ b/tests/key-store.test.ts @@ -0,0 +1,127 @@ +import { jwtVerify } from "jose" +import { describe, expect, it } from "vitest" +import { createKeyStore, memoryPersistence } from "../src" +import { clock, fakeFetch, JWKS_URL, nowSec, type SigningKey, serviceClaims, sign, signingKey } from "./idp" + +async function setup(initial: SigningKey[]) { + let published = initial + let online = true + const time = clock() + const persistence = memoryPersistence() + const { fetch, requests } = fakeFetch(() => + online ? Response.json({ keys: published.map((key) => key.publicJwk) }) : new Response(null, { status: 503 }) + ) + const make = () => createKeyStore({ jwksUrl: JWKS_URL, persistence, fetch, now: time.now, onError: () => {} }) + return { + time, + persistence, + requests, + make, + publish: (keys: SigningKey[]) => { + published = keys + }, + setOnline: (value: boolean) => { + online = value + }, + } +} + +const verify = (token: string, store: { getKey: Parameters[1] }) => + jwtVerify(token, store.getKey as never, { algorithms: ["EdDSA"] }) + +describe("createKeyStore", () => { + it("fetches, persists and resolves keys by kid", async () => { + const first = await signingKey("k1") + const env = await setup([first]) + const store = env.make() + await store.start() + store.stop() + expect(store.status()).toMatchObject({ kids: ["k1"] }) + expect(await env.persistence.load(`jwks:${JWKS_URL}`)).toContain("k1") + await expect(verify(await sign(first, serviceClaims()), store)).resolves.toBeTruthy() + }) + + it("refetches immediately for a token newer than the last fetch, and fails closed on an unknown kid", async () => { + const first = await signingKey("k1") + const second = await signingKey("k2") + const env = await setup([first]) + const store = env.make() + await store.start() + store.stop() + env.publish([first, second]) + env.time.advance(2_000) + // Issued after the last fetch: a key the IdP created since then. + const fresh = await sign(second, serviceClaims({ iat: nowSec() + 2 })) + await expect(verify(fresh, store)).resolves.toBeTruthy() + expect(env.requests).toHaveLength(2) + + // Within 30 s of the last fetch, an old token with an unknown kid triggers no fetch at all. + const stranger = await signingKey("unknown") + const early = await sign(stranger, serviceClaims({ iat: nowSec() - 60 })) + await expect(verify(early, store)).rejects.toThrow() + expect(env.requests).toHaveLength(2) + env.time.advance(30_000) + const old = await sign(stranger, serviceClaims({ iat: nowSec() - 60 })) + await expect(verify(old, store)).rejects.toMatchObject({ code: "ERR_JWKS_NO_MATCHING_KEY" }) + await expect(verify(old, store)).rejects.toMatchObject({ code: "ERR_JWKS_NO_MATCHING_KEY" }) + // Old tokens with unknown kids refetch at most once per 30 s. + expect(env.requests).toHaveLength(3) + }) + + it("refetches for a token issued in the same second as the last fetch", async () => { + const first = await signingKey("k1") + const second = await signingKey("k2") + const env = await setup([first]) + const store = env.make() + await store.start() + store.stop() + const fetchSecond = Math.floor(env.time.now() / 1000) + env.publish([first, second]) + env.time.advance(1_000) + const token = await sign(second, serviceClaims({ iat: fetchSecond })) + await expect(verify(token, store)).resolves.toBeTruthy() + expect(env.requests).toHaveLength(2) + }) + + it("ignores persisted keys dated in the future", async () => { + const first = await signingKey("k1") + const env = await setup([first]) + await env.persistence.save( + `jwks:${JWKS_URL}`, + JSON.stringify({ fetchedAt: env.time.now() + 86_400_000, jwks: { keys: [first.publicJwk] } }) + ) + env.setOnline(false) + const store = env.make() + await store.start() + store.stop() + expect(store.status().fetchedAt).toBeNull() + }) + + it("keeps the persisted keys through an outage until they are a day old", async () => { + const first = await signingKey("k1") + const env = await setup([first]) + const warm = env.make() + await warm.start() + warm.stop() + + env.setOnline(false) + const restarted = env.make() + await restarted.start() + restarted.stop() + const token = await sign(first, serviceClaims()) + await expect(verify(token, restarted)).resolves.toBeTruthy() + env.time.advance(86_400_001) + await expect(verify(token, restarted)).rejects.toMatchObject({ code: "keys_unavailable" }) + }) + + it("fails closed on a cold start with the IdP unreachable", async () => { + const first = await signingKey("k1") + const env = await setup([first]) + env.setOnline(false) + const store = env.make() + await store.start() + store.stop() + await expect(verify(await sign(first, serviceClaims()), store)).rejects.toMatchObject({ code: "keys_unavailable" }) + expect(store.status()).toEqual({ fetchedAt: null, kids: [] }) + }) +}) diff --git a/tests/snapshot-client.test.ts b/tests/snapshot-client.test.ts new file mode 100644 index 0000000..792c834 --- /dev/null +++ b/tests/snapshot-client.test.ts @@ -0,0 +1,202 @@ +import { describe, expect, it } from "vitest" +import { type Actor, accessSnapshot, memoryPersistence, type TokenSource } from "../src" +import { fixture } from "./fixtures" +import { clock, fakeFetch } from "./idp" + +const URL = "http://auth.internal/api/internal/access-snapshot" +const BODY = fixture("snapshot-v1.json") +const ETAG = '"sha256-aaaa"' +const moderator: Actor = { kind: "telegram", client: "telegram-bot", telegramId: "123456789", sub: "usr_moderator" } + +function tokenSource() { + let n = 0 + let invalidated = 0 + const tokens: TokenSource = { + getToken: async () => `token-${n}`, + invalidate: () => { + invalidated++ + n++ + }, + } + return { tokens, invalidated: () => invalidated } +} + +type Reply = (request: Request) => Response | Promise + +function setup(reply: Reply, persistence = memoryPersistence()) { + const time = clock(Date.parse("2026-10-09T10:20:00.000Z")) + const { tokens, invalidated } = tokenSource() + const errors: unknown[] = [] + const { fetch, requests } = fakeFetch(reply) + const client = accessSnapshot({ + url: URL, + tokens, + projection: "backend", + persistence, + fetch, + now: time.now, + onError: (error) => errors.push(error), + }) + return { client, time, requests, errors, persistence, invalidated } +} + +const ok = () => new Response(BODY, { status: 200, headers: { ETag: ETAG } }) + +describe("accessSnapshot", () => { + it("pulls with a bearer token, swaps in the snapshot and answers permission checks", async () => { + const { client, requests } = setup(ok) + await client.start() + client.stop() + expect(requests[0]?.headers.get("authorization")).toBe("Bearer token-0") + expect(requests[0]?.headers.has("if-none-match")).toBe(false) + expect(client.status()).toMatchObject({ fresh: true, generation: 4813, subjects: 5 }) + expect(client.has(moderator, "tg:moderate")).toBe(true) + expect(client.has(moderator, "tg:moderate:global")).toBe(false) + expect(client.subjectByTelegramId("555")?.sub).toBe("usr_direttivo") + }) + + it("denies service actors, actors without sub, and expired permission paths", async () => { + const { client, time } = setup(ok) + await client.start() + client.stop() + expect(client.has({ kind: "service", client: "telegram-bot" }, "tg:moderate")).toBe(false) + expect(client.has({ ...moderator, sub: null }, "tg:moderate")).toBe(false) + const direttivo: Actor = { kind: "user", client: "admin-dashboard", sub: "usr_direttivo", telegramId: "555" } + expect(client.has(direttivo, "tg:moderate:global")).toBe(true) + time.advance(55 * 60_000) + expect(client.has(direttivo, "tg:moderate:global")).toBe(false) + expect(client.has(direttivo, "tg:audit:read")).toBe(true) + }) + + it("revalidates with the ETag and counts a 304 as a sync", async () => { + let calls = 0 + const { client, time, requests } = setup(() => (++calls === 1 ? ok() : new Response(null, { status: 304 }))) + await client.start() + client.stop() + time.advance(50 * 60_000) + await client.refresh() + expect(requests[1]?.headers.get("if-none-match")).toBe(ETAG) + time.advance(50 * 60_000) + expect(client.status().fresh).toBe(true) + expect(client.has(moderator, "tg:moderate")).toBe(true) + }) + + it("fails closed once the last sync is older than MAX_STALE", async () => { + let online = true + const { client, time } = setup(() => (online ? ok() : new Response(null, { status: 503 }))) + await client.start() + client.stop() + online = false + time.advance(60 * 60_000) + await expect(client.refresh()).rejects.toMatchObject({ code: "snapshot_unavailable" }) + expect(client.has(moderator, "tg:moderate")).toBe(true) + time.advance(1) + expect(client.has(moderator, "tg:moderate")).toBe(false) + expect(client.subjectBySub("usr_moderator")).toBeUndefined() + expect(client.current()?.bySub("usr_moderator")).toBeDefined() + }) + + it("renews the token once after a 401", async () => { + let calls = 0 + const { client, invalidated, requests } = setup((request) => + ++calls === 1 && request.headers.get("authorization") === "Bearer token-0" + ? Response.json({ error: "invalid_token" }, { status: 401 }) + : ok() + ) + await client.refresh() + expect(invalidated()).toBe(1) + expect(requests[1]?.headers.get("authorization")).toBe("Bearer token-1") + expect(client.status().fresh).toBe(true) + }) + + it("keeps the previous snapshot when a new one is invalid", async () => { + let calls = 0 + const { client } = setup(() => (++calls === 1 ? ok() : new Response(fixture("invalid/duplicate-sub.json")))) + await client.refresh() + await expect(client.refresh()).rejects.toMatchObject({ code: "snapshot_invalid" }) + expect(client.status().generation).toBe(4813) + expect(client.has(moderator, "tg:moderate")).toBe(true) + }) + + it("resumes from the persisted snapshot after a restart during an outage", async () => { + const persistence = memoryPersistence() + const first = setup(ok, persistence) + await first.client.start() + first.client.stop() + + const restarted = setup(() => new Response(null, { status: 503 }), persistence) + await restarted.client.start() + restarted.client.stop() + expect(restarted.errors).toHaveLength(1) + expect(restarted.client.has(moderator, "tg:moderate")).toBe(true) + restarted.time.advance(60 * 60_000 + 1) + expect(restarted.client.has(moderator, "tg:moderate")).toBe(false) + }) + + it("ignores a persisted snapshot dated in the future", async () => { + const persistence = memoryPersistence() + await persistence.save( + "snapshot:backend", + JSON.stringify({ body: BODY, etag: ETAG, lastSyncAt: Date.parse("2027-01-01T00:00:00.000Z") }) + ) + const { client } = setup(() => new Response(null, { status: 503 }), persistence) + await client.start() + client.stop() + expect(client.status().lastSyncAt).toBeNull() + expect(client.has(moderator, "tg:moderate")).toBe(false) + }) + + it("abandons a pull that hangs and keeps the previous snapshot", async () => { + let calls = 0 + const time = clock() + const { fetch } = fakeFetch((request) => + ++calls === 1 + ? ok() + : new Promise((_, reject) => { + request.signal.addEventListener("abort", () => reject(request.signal.reason)) + }) + ) + const client = accessSnapshot({ + url: URL, + tokens: tokenSource().tokens, + projection: "backend", + fetch, + now: time.now, + timeoutMs: 20, + }) + await client.refresh() + await expect(client.refresh()).rejects.toMatchObject({ code: "snapshot_unavailable" }) + expect(client.has(moderator, "tg:moderate")).toBe(true) + }) + + it("fails closed on a cold start without a stored snapshot", async () => { + const { client } = setup(() => new Response(null, { status: 503 })) + await client.start() + client.stop() + expect(client.status()).toEqual({ lastSyncAt: null, fresh: false, generation: null, subjects: 0 }) + expect(client.has(moderator, "tg:moderate")).toBe(false) + }) + + it("runs one pull at a time and queues a single follow-up for triggers during a pull", async () => { + const pending: Array<() => void> = [] + const { client, requests } = setup( + () => + new Promise((resolve) => { + pending.push(() => resolve(ok())) + }) + ) + const first = client.refresh() + const second = client.refresh() + const third = client.refresh() + expect(second).toBe(third) + await Promise.resolve() + expect(requests).toHaveLength(1) + pending.shift()?.() + await first + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(requests).toHaveLength(2) + pending.shift()?.() + await second + expect(requests).toHaveLength(2) + }) +}) diff --git a/tests/token-source.test.ts b/tests/token-source.test.ts new file mode 100644 index 0000000..151a30d --- /dev/null +++ b/tests/token-source.test.ts @@ -0,0 +1,149 @@ +import { createLocalJWKSet, jwtVerify } from "jose" +import { describe, expect, it } from "vitest" +import { AuthKitError, CLIENT_ASSERTION_TYPE, serviceTokenSource } from "../src" +import { BACKEND, clock, fakeFetch, ISSUER, signingKey } from "./idp" + +const TOKEN_URL = "http://auth.internal/api/auth/oauth2/token" +const TOKEN_AUDIENCE = `${ISSUER}/oauth2/token` + +async function setup(respond: (count: number) => Response, ratio?: number) { + const key = await signingKey("bot-key") + const time = clock() + let count = 0 + const errors: unknown[] = [] + const { fetch, requests } = fakeFetch(() => respond(++count)) + const tokens = serviceTokenSource({ + clientId: "telegram-bot", + privateKey: key.privateJwk, + tokenUrl: TOKEN_URL, + tokenAudience: TOKEN_AUDIENCE, + resource: BACKEND, + scopes: ["backend:tg:read", "backend:tg:ingest"], + fetch, + now: time.now, + onError: (error) => errors.push(error), + ...(ratio === undefined ? {} : { refreshRatio: ratio }), + }) + return { key, time, tokens, requests, errors, count: () => count } +} + +/** Lets a background renewal finish. */ +const settle = () => new Promise((resolve) => setTimeout(resolve, 0)) + +const issued = (n: number) => + Response.json({ access_token: `token-${n}`, token_type: "Bearer", expires_in: 3600, scope: "backend:tg:read" }) + +describe("serviceTokenSource", () => { + it("requests client credentials with a private_key_jwt assertion and one resource", async () => { + const { key, tokens, requests } = await setup(issued) + expect(await tokens.getToken()).toBe("token-1") + const request = requests[0] + expect(request?.url).toBe(TOKEN_URL) + expect(request?.method).toBe("POST") + const form = new URLSearchParams(await request?.text()) + expect(form.get("grant_type")).toBe("client_credentials") + expect(form.get("client_id")).toBe("telegram-bot") + expect(form.get("client_assertion_type")).toBe(CLIENT_ASSERTION_TYPE) + expect(form.getAll("resource")).toEqual([BACKEND]) + expect(form.get("scope")).toBe("backend:tg:read backend:tg:ingest") + const { payload, protectedHeader } = await jwtVerify( + form.get("client_assertion") ?? "", + createLocalJWKSet({ keys: [key.publicJwk] }), + { issuer: "telegram-bot", subject: "telegram-bot", audience: TOKEN_AUDIENCE } + ) + expect(protectedHeader).toMatchObject({ alg: "EdDSA", kid: "bot-key" }) + expect(Number(payload.exp) - Number(payload.iat)).toBe(60) + expect(payload.jti).toEqual(expect.any(String)) + }) + + it("caches the token, shares one request between concurrent callers and renews at half the lifetime", async () => { + const { tokens, time, count } = await setup(issued) + expect(await Promise.all([tokens.getToken(), tokens.getToken(), tokens.getToken()])).toEqual([ + "token-1", + "token-1", + "token-1", + ]) + expect(count()).toBe(1) + time.advance(1_799_000) + expect(await tokens.getToken()).toBe("token-1") + expect(count()).toBe(1) + time.advance(2_000) + // Past the renewal point the valid token is returned at once and a new one is fetched. + expect(await tokens.getToken()).toBe("token-1") + await settle() + expect(count()).toBe(2) + expect(await tokens.getToken()).toBe("token-2") + }) + + it("keeps serving a cached token while renewal fails, then fails once it expires", async () => { + const { tokens, time, errors, count } = await setup((n) => + n === 1 + ? issued(n) + : Response.json({ error: "server_error", error_description: "secret detail" }, { status: 500 }) + ) + await tokens.getToken() + time.advance(2_000_000) + expect(await tokens.getToken()).toBe("token-1") + await settle() + expect(errors).toHaveLength(1) + expect((errors[0] as Error).message).not.toContain("secret detail") + // Inside the backoff window no new request is made. + expect(await tokens.getToken()).toBe("token-1") + expect(count()).toBe(2) + time.advance(1_600_000) + await expect(tokens.getToken()).rejects.toMatchObject({ code: "token_request_failed" }) + }) + + it("does not retry a failed request inside the backoff window", async () => { + const { tokens, time, count } = await setup(() => new Response(null, { status: 503 })) + await expect(tokens.getToken()).rejects.toBeInstanceOf(AuthKitError) + await expect(tokens.getToken()).rejects.toMatchObject({ code: "token_request_failed" }) + expect(count()).toBe(1) + time.advance(5_000) + await expect(tokens.getToken()).rejects.toMatchObject({ code: "token_request_failed" }) + expect(count()).toBe(2) + }) + + it("abandons a token request that hangs", async () => { + const key = await signingKey("bot-key") + const { fetch } = fakeFetch( + (request) => + new Promise((_, reject) => { + request.signal.addEventListener("abort", () => reject(request.signal.reason)) + }) + ) + const tokens = serviceTokenSource({ + clientId: "telegram-bot", + privateKey: key.privateJwk, + tokenUrl: TOKEN_URL, + tokenAudience: TOKEN_AUDIENCE, + resource: BACKEND, + scopes: [], + fetch, + timeoutMs: 20, + }) + await expect(tokens.getToken()).rejects.toMatchObject({ code: "token_request_failed" }) + }) + + it("fetches a new token after invalidate()", async () => { + const { tokens } = await setup(issued) + await tokens.getToken() + tokens.invalidate() + expect(await tokens.getToken()).toBe("token-2") + }) + + it("rejects unusable responses and keys", async () => { + const { tokens } = await setup(() => Response.json({ access_token: "x", token_type: "DPoP", expires_in: 60 })) + await expect(tokens.getToken()).rejects.toMatchObject({ code: "token_request_failed" }) + const key = await signingKey("k") + const base = { + clientId: "c", + tokenUrl: TOKEN_URL, + tokenAudience: TOKEN_AUDIENCE, + resource: BACKEND, + scopes: [], + } + expect(() => serviceTokenSource({ ...base, privateKey: key.publicJwk })).toThrow(TypeError) + expect(() => serviceTokenSource({ ...base, privateKey: { ...key.privateJwk, alg: "RS256" } })).toThrow(TypeError) + }) +}) diff --git a/tests/verify.test.ts b/tests/verify.test.ts new file mode 100644 index 0000000..ca26437 --- /dev/null +++ b/tests/verify.test.ts @@ -0,0 +1,128 @@ +import type { JWTPayload } from "jose" +import { describe, expect, it } from "vitest" +import { createKeyStore, parseAccessSnapshot, resolveActor, type VerifiedToken, verifyAccessToken } from "../src" +import { fixture } from "./fixtures" +import { + BACKEND, + fakeFetch, + ISSUER, + JWKS_URL, + nowSec, + serviceClaims, + sign, + signingKey, + userClaims, + without, +} from "./idp" + +const key = await signingKey("k1") +const store = createKeyStore({ + jwksUrl: JWKS_URL, + fetch: fakeFetch(() => Response.json({ keys: [key.publicJwk] })).fetch, +}) +await store.start() +store.stop() + +const options = { issuer: ISSUER, audience: BACKEND, keyStore: store } +const verify = async (claims: JWTPayload, header?: Parameters[2]) => + verifyAccessToken(await sign(key, claims, header), options) + +describe("verifyAccessToken", () => { + it("accepts a service token and a user token whose aud is an array", async () => { + const service = await verify(serviceClaims()) + expect(service).toMatchObject({ kind: "service", clientId: "telegram-bot", sub: "telegram-bot" }) + expect([...service.scopes]).toEqual(["backend:tg:read", "backend:tg:act-as"]) + expect(await verify(userClaims())).toMatchObject({ + kind: "user", + clientId: "admin-dashboard", + sub: "usr_moderator", + }) + }) + + it.each([ + ["another issuer", serviceClaims({ iss: "https://evil.example/api/auth" }), undefined], + ["another audience", serviceClaims({ aud: "https://auth.example.org/api/internal" }), undefined], + ["an expired token", serviceClaims({ iat: nowSec() - 7200, exp: nowSec() - 3600 }), undefined], + ["a token older than 2 h", serviceClaims({ iat: nowSec() - 3 * 3600, exp: nowSec() + 60 }), undefined], + ["a missing jti", without(serviceClaims(), "jti"), undefined], + ["a missing subject type", without(serviceClaims(), "pn_subject_type"), undefined], + ["a client token whose sub is not the client", serviceClaims({ sub: "usr_x" }), undefined], + ["a user token whose sub is the client", userClaims({ sub: "admin-dashboard" }), undefined], + ["an unknown subject type", serviceClaims({ pn_subject_type: "robot" }), undefined], + ["an ID token", userClaims(), { typ: "JWT" }], + ["a security event token", serviceClaims(), { typ: "secevent+jwt" }], + ])("rejects %s", async (_name, claims, header) => { + await expect(verify(claims, header)).rejects.toMatchObject({ code: "token_invalid" }) + }) + + it("rejects algorithms other than EdDSA", async () => { + const es256 = await signingKey("k1", "ES256") + await expect(verifyAccessToken(await sign(es256, serviceClaims()), options)).rejects.toMatchObject({ + code: "token_invalid", + }) + }) + + it("reports missing keys as keys_unavailable rather than an invalid token", async () => { + const offline = createKeyStore({ + jwksUrl: JWKS_URL, + fetch: fakeFetch(() => new Response(null, { status: 503 })).fetch, + onError: () => {}, + }) + await expect( + verifyAccessToken(await sign(key, serviceClaims()), { ...options, keyStore: offline }) + ).rejects.toMatchObject({ code: "keys_unavailable" }) + }) +}) + +describe("resolveActor", () => { + const snapshot = parseAccessSnapshot(fixture("snapshot-v1.json"), "backend") + const token = (kind: "service" | "user", scopes: string[], sub = "telegram-bot"): VerifiedToken => ({ + kind, + clientId: kind === "service" ? sub : "admin-dashboard", + sub, + scopes: new Set(scopes), + claims: {}, + }) + const actAsScope = "backend:tg:act-as" + + it("returns the service itself without an actor header", () => { + expect(resolveActor(token("service", []), { actAsScope, snapshot })).toEqual({ + kind: "service", + client: "telegram-bot", + }) + }) + + it("fills in a user's Telegram ID from the snapshot", () => { + expect(resolveActor(token("user", [], "usr_moderator"), { actAsScope, snapshot })).toEqual({ + kind: "user", + client: "admin-dashboard", + sub: "usr_moderator", + telegramId: "123456789", + }) + expect(resolveActor(token("user", [], "usr_moderator"), { actAsScope, snapshot: null })).toMatchObject({ + telegramId: null, + }) + }) + + it("lets a service with the act-as scope assert a Telegram actor", () => { + const service = token("service", [actAsScope]) + expect(resolveActor(service, { actAsScope, snapshot, actorHeader: "telegram:123456789" })).toEqual({ + kind: "telegram", + client: "telegram-bot", + telegramId: "123456789", + sub: "usr_moderator", + }) + expect(resolveActor(service, { actAsScope, snapshot, actorHeader: "telegram:42" })).toMatchObject({ sub: null }) + }) + + it.each([ + ["a service without the act-as scope", token("service", ["backend:tg:read"]), "telegram:1"], + ["a user token", token("user", [actAsScope], "usr_moderator"), "telegram:1"], + ["a malformed header", token("service", [actAsScope]), "telegram:abc"], + ["an empty header", token("service", [actAsScope]), ""], + ])("forbids an actor header from %s", (_name, verified, actorHeader) => { + expect(() => resolveActor(verified, { actAsScope, snapshot, actorHeader })).toThrowError( + expect.objectContaining({ code: "actor_forbidden" }) + ) + }) +}) From 4b95986d778db6c1ee5232c6b961211051a900f1 Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Fri, 9 Oct 2026 19:33:20 +0000 Subject: [PATCH 2/3] fix: reject snapshot 304 responses without a cached ETag --- src/snapshot/client.ts | 4 +++- tests/snapshot-client.test.ts | 29 +++++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/src/snapshot/client.ts b/src/snapshot/client.ts index d9a1408..3e5f3b0 100644 --- a/src/snapshot/client.ts +++ b/src/snapshot/client.ts @@ -103,7 +103,9 @@ export function accessSnapshot(options: AccessSnapshotOptions): AccessSnapshotCl response = await request(await options.tokens.getToken()) } if (response.status === 304) { - if (!state) throw new AuthKitError("snapshot_unavailable", "Snapshot endpoint answered 304 without a snapshot") + if (!state?.etag) { + throw new AuthKitError("snapshot_unavailable", "Snapshot endpoint answered 304 without a snapshot and ETag") + } lastSyncAt = now() await persist() return diff --git a/tests/snapshot-client.test.ts b/tests/snapshot-client.test.ts index 792c834..e74849e 100644 --- a/tests/snapshot-client.test.ts +++ b/tests/snapshot-client.test.ts @@ -81,6 +81,35 @@ describe("accessSnapshot", () => { expect(client.has(moderator, "tg:moderate")).toBe(true) }) + it.each([null, ""])("rejects a 304 when the snapshot ETag is %j", async (etag) => { + let calls = 0 + const { client, time, requests, persistence } = setup(() => + ++calls === 1 + ? new Response(BODY, { headers: etag === null ? {} : { ETag: etag } }) + : new Response(null, { status: 304 }) + ) + await client.refresh() + const index = client.current() + const lastSyncAt = client.status().lastSyncAt + const stored = await persistence.load("snapshot:backend") + time.advance(60 * 60_000 + 1) + + await expect(client.refresh()).rejects.toMatchObject({ code: "snapshot_unavailable" }) + + expect(requests[1]?.headers.has("if-none-match")).toBe(false) + expect(client.current()).toBe(index) + expect(client.status()).toMatchObject({ lastSyncAt, fresh: false }) + expect(client.has(moderator, "tg:moderate")).toBe(false) + expect(await persistence.load("snapshot:backend")).toBe(stored) + }) + + it("rejects a 304 without a snapshot", async () => { + const { client, persistence } = setup(() => new Response(null, { status: 304 })) + await expect(client.refresh()).rejects.toMatchObject({ code: "snapshot_unavailable" }) + expect(client.status()).toEqual({ lastSyncAt: null, fresh: false, generation: null, subjects: 0 }) + expect(await persistence.load("snapshot:backend")).toBeNull() + }) + it("fails closed once the last sync is older than MAX_STALE", async () => { let online = true const { client, time } = setup(() => (online ? ok() : new Response(null, { status: 503 }))) From 1b8b779edf061606e935d75f20a9adc78bdca38e Mon Sep 17 00:00:00 2001 From: Tommaso Morganti Date: Sat, 10 Oct 2026 00:18:26 +0200 Subject: [PATCH 3/3] fix(test): strenghten completion check of tests --- tests/token-source.test.ts | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/tests/token-source.test.ts b/tests/token-source.test.ts index 151a30d..5636e4f 100644 --- a/tests/token-source.test.ts +++ b/tests/token-source.test.ts @@ -27,8 +27,14 @@ async function setup(respond: (count: number) => Response, ratio?: number) { return { key, time, tokens, requests, errors, count: () => count } } -/** Lets a background renewal finish. */ -const settle = () => new Promise((resolve) => setTimeout(resolve, 0)) +/** Waits for a background renewal to reach the given state. */ +async function until(condition: () => boolean | Promise) { + for (let i = 0; i < 200; i++) { + if (await condition()) return + await new Promise((resolve) => setTimeout(resolve, 1)) + } + throw new Error("condition not met") +} const issued = (n: number) => Response.json({ access_token: `token-${n}`, token_type: "Bearer", expires_in: 3600, scope: "backend:tg:read" }) @@ -70,9 +76,8 @@ describe("serviceTokenSource", () => { time.advance(2_000) // Past the renewal point the valid token is returned at once and a new one is fetched. expect(await tokens.getToken()).toBe("token-1") - await settle() + await until(async () => (await tokens.getToken()) === "token-2") expect(count()).toBe(2) - expect(await tokens.getToken()).toBe("token-2") }) it("keeps serving a cached token while renewal fails, then fails once it expires", async () => { @@ -84,7 +89,7 @@ describe("serviceTokenSource", () => { await tokens.getToken() time.advance(2_000_000) expect(await tokens.getToken()).toBe("token-1") - await settle() + await until(() => errors.length === 1) expect(errors).toHaveLength(1) expect((errors[0] as Error).message).not.toContain("secret detail") // Inside the backoff window no new request is made.