diff --git a/apps/journal/app/jobs/consumed-jti-sweep.ts b/apps/journal/app/jobs/consumed-jti-sweep.ts new file mode 100644 index 0000000..1b8a7c0 --- /dev/null +++ b/apps/journal/app/jobs/consumed-jti-sweep.ts @@ -0,0 +1,31 @@ +import type { JobDefinition } from "@trails-cool/jobs"; +import { lt } from "drizzle-orm"; +import { consumedJwtJti } from "@trails-cool/db/schema/journal"; +import { getDb } from "../lib/db.ts"; +import { logger } from "../lib/logger.server.ts"; + +/** + * Daily cleanup for the JWT replay-protection table. Each row was + * inserted by `verifyRouteToken` to mark a token as consumed; once + * the token's `exp` claim has passed, the row is no longer useful + * (the JWT itself would fail signature verification before reaching + * the consume step). Bound the table to keep it tiny. + * + * See planner-audit #2 Phase B. + */ +export const consumedJtiSweepJob: JobDefinition = { + name: "consumed-jti-sweep", + cron: "45 3 * * *", // daily at 03:45 UTC (offset from notifications-purge) + retryLimit: 1, + expireInSeconds: 60, + async handler() { + const db = getDb(); + const result = await db + .delete(consumedJwtJti) + .where(lt(consumedJwtJti.expiresAt, new Date())) + .returning({ jti: consumedJwtJti.jti }); + const purged = result.length; + logger.info({ purged }, "consumed-jti-sweep"); + return { purged }; + }, +}; diff --git a/apps/journal/app/lib/jwt.server.ts b/apps/journal/app/lib/jwt.server.ts index 333b8fb..b7e1637 100644 --- a/apps/journal/app/lib/jwt.server.ts +++ b/apps/journal/app/lib/jwt.server.ts @@ -1,5 +1,8 @@ import { SignJWT, jwtVerify } from "jose"; +import { randomUUID } from "node:crypto"; import { getOrigin, requireSecret } from "./config.server.ts"; +import { getDb } from "./db.ts"; +import { consumedJwtJti } from "@trails-cool/db/schema/journal"; const JWT_SECRET = new TextEncoder().encode( requireSecret("JWT_SECRET", "dev-jwt-secret-change-in-production"), @@ -13,14 +16,57 @@ export async function createRouteToken(routeId: string, permissions: string[] = .setIssuer(ISSUER) .setExpirationTime("7d") .setIssuedAt() + .setJti(randomUUID()) .sign(JWT_SECRET); } +export class TokenAlreadyConsumedError extends Error { + constructor() { + super("Token already consumed"); + this.name = "TokenAlreadyConsumedError"; + } +} + +/** + * Verify a route token AND atomically consume it. Subsequent calls + * with the same token throw `TokenAlreadyConsumedError`. + * + * The consume step is `INSERT … ON CONFLICT DO NOTHING RETURNING jti`. + * Postgres serializes the insert against concurrent attempts, so + * exactly one caller observes the returned row — the rest see an + * empty result and reject. + * + * Tokens minted before this PR (no `jti` claim) are rejected outright, + * which is the right behavior: pre-existing tokens are already in + * planner-session DB rows and would be replayable if accepted. The + * user re-saves by going back to the journal for a fresh token. + */ export async function verifyRouteToken(token: string): Promise<{ routeId: string; permissions: string[] }> { const { payload } = await jwtVerify(token, JWT_SECRET, { issuer: ISSUER, }); + if (typeof payload.jti !== "string" || !payload.jti) { + throw new TokenAlreadyConsumedError(); + } + if (typeof payload.exp !== "number") { + throw new TokenAlreadyConsumedError(); + } + + const db = getDb(); + const inserted = await db + .insert(consumedJwtJti) + .values({ + jti: payload.jti, + expiresAt: new Date(payload.exp * 1000), + }) + .onConflictDoNothing() + .returning({ jti: consumedJwtJti.jti }); + + if (inserted.length === 0) { + throw new TokenAlreadyConsumedError(); + } + return { routeId: payload.route_id as string, permissions: payload.permissions as string[], diff --git a/apps/journal/server.ts b/apps/journal/server.ts index 901db0f..a140502 100644 --- a/apps/journal/server.ts +++ b/apps/journal/server.ts @@ -176,8 +176,9 @@ server.listen(port, async () => { const { komootBulkImportJob } = await import("./app/jobs/komoot-bulk-import.ts"); const { importBatchesSweepJob } = await import("./app/jobs/import-batches-sweep.ts"); const { sendWelcomeEmailJob } = await import("./app/jobs/send-welcome-email.ts"); + const { consumedJtiSweepJob } = await import("./app/jobs/consumed-jti-sweep.ts"); // eslint-disable-next-line @typescript-eslint/no-explicit-any - jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob as any, importBatchesSweepJob, sendWelcomeEmailJob); + jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob as any, importBatchesSweepJob, sendWelcomeEmailJob, consumedJtiSweepJob); const boss = createBoss(getDatabaseUrl()); await startWorker(boss, jobs); diff --git a/e2e/integration.test.ts b/e2e/integration.test.ts index 4bad37c..4fdb784 100644 --- a/e2e/integration.test.ts +++ b/e2e/integration.test.ts @@ -71,6 +71,28 @@ test.describe("Integration: Planner callback → geometry stored", () => { }); expect(resp.status()).toBe(401); }); + + test("token is single-use — second submit is rejected (Phase B replay guard)", async ({ request }) => { + const seedResp = await request.post(`${JOURNAL}/api/e2e/seed`); + const { routeId, token } = await seedResp.json() as { routeId: string; token: string }; + + // First save succeeds. + const first = await request.post(`${JOURNAL}/api/routes/${routeId}/callback`, { + headers: { Authorization: `Bearer ${token}` }, + data: { gpx: VALID_GPX }, + }); + expect(first.status()).toBe(200); + + // Second attempt with the *same* token must fail — the jti has + // been consumed. + const second = await request.post(`${JOURNAL}/api/routes/${routeId}/callback`, { + headers: { Authorization: `Bearer ${token}` }, + data: { gpx: VALID_GPX }, + }); + expect(second.status()).toBe(401); + const body = await second.json() as { error: string }; + expect(body.error).toMatch(/consumed|already/i); + }); }); test.describe("Integration: Journal ↔ Planner handoff", () => { diff --git a/packages/db/src/schema/journal.ts b/packages/db/src/schema/journal.ts index 20c0357..ca438ec 100644 --- a/packages/db/src/schema/journal.ts +++ b/packages/db/src/schema/journal.ts @@ -356,3 +356,17 @@ export const notifications = journalSchema.table("notifications", { .on(t.recipientUserId, t.type, t.subjectId) .where(sql`${t.subjectId} IS NOT NULL`), })); + +// Single-use JWT enforcement for callback tokens (planner → journal +// save flow). Each token carries a `jti` claim; the verifier inserts +// into this table on first use. A second attempt hits the PK conflict +// and the verifier rejects the token. Rows are swept after `expires_at`. +// See spec / planner-audit #2 Phase B. +export const consumedJwtJti = journalSchema.table("consumed_jwt_jti", { + jti: text("jti").primaryKey(), + consumedAt: timestamp("consumed_at", { withTimezone: true }).notNull().defaultNow(), + expiresAt: timestamp("expires_at", { withTimezone: true }).notNull(), +}, (t) => ({ + // Sweep runs `DELETE WHERE expires_at < now()` on a daily schedule. + expiresAtIdx: index("consumed_jwt_jti_expires_at_idx").on(t.expiresAt), +}));