diff --git a/Realism/app/api/jobs/webhook/route.ts b/Realism/app/api/jobs/webhook/route.ts index e72ad0c..134fef8 100644 --- a/Realism/app/api/jobs/webhook/route.ts +++ b/Realism/app/api/jobs/webhook/route.ts @@ -1,29 +1,65 @@ -import { NextRequest, NextResponse } from 'next/server' -import { inngest } from '@/lib/inngest' - -export async function POST(req: NextRequest) { - let jobId: string | undefined - - try { - const body = await req.json() - jobId = body.jobId - } catch { - return NextResponse.json({ error: 'Invalid body' }, { status: 400 }) - } - - if (!jobId) { - return NextResponse.json({ error: 'jobId required' }, { status: 400 }) - } - - try { - await inngest.send({ - name: 'job/webhook', - data: { jobId }, - }) - } catch (err) { - console.error('[webhook] Failed to send Inngest event:', err) - } - - // Always 200 to prevent QStash retries - return NextResponse.json({ ok: true }) -} +import { NextRequest, NextResponse } from 'next/server' +import { inngest } from '@/lib/inngest' +import { Receiver } from '@upstash/qstash' +import { checkAndLockMessage, unlockMessage } from '@/lib/jobs' + +// SECURITY FIX: Initialize QStash receiver for signature verification +const receiver = new Receiver({ + currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY || '', + nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY || '', +}) + +export async function POST(req: NextRequest) { + // SECURITY FIX: Verify provider signatures to reject unsigned/unauthenticated requests. + const signature = req.headers.get('Upstash-Signature') + const messageId = req.headers.get('Upstash-Message-Id') + + if (!signature || !messageId) { + return NextResponse.json({ error: 'Missing QStash signature or message ID' }, { status: 401 }) + } + + const bodyText = await req.text() + + try { + const isValid = await receiver.verify({ signature, body: bodyText }) + if (!isValid) throw new Error("Invalid signature") + } catch (err) { + return NextResponse.json({ error: 'Unauthorized: Invalid signature' }, { status: 401 }) + } + + let jobId: string | undefined + + try { + const body = JSON.parse(bodyText) + jobId = body.jobId + } catch { + return NextResponse.json({ error: 'Invalid JSON body' }, { status: 400 }) + } + + if (!jobId) { + return NextResponse.json({ error: 'jobId required' }, { status: 400 }) + } + + // SECURITY FIX: Persist a one-time delivery ID (Idempotency Key) with SETNX. + // This prevents the same webhook request from being processed multiple times. + const isDuplicate = await checkAndLockMessage(messageId) + if (isDuplicate) { + return NextResponse.json({ ok: true, note: 'Message already processed' }) + } + + try { + await inngest.send({ + name: 'job/webhook', + data: { jobId }, + }) + } catch (err) { + console.error('[webhook] Failed to send Inngest event:', err) + + // SECURITY FIX: Do NOT swallow errors. If delivery fails, we release the idempotency lock + // and return a 500 status code so the upstream scheduler can properly retry the failed delivery. + await unlockMessage(messageId) + return NextResponse.json({ error: 'Event delivery failed' }, { status: 500 }) + } + + return NextResponse.json({ ok: true }) +} \ No newline at end of file diff --git a/Realism/lib/jobs.ts b/Realism/lib/jobs.ts index 9f6c18c..925c437 100644 --- a/Realism/lib/jobs.ts +++ b/Realism/lib/jobs.ts @@ -1,85 +1,123 @@ -import { randomUUID } from 'crypto' -import { setJob, getJob, updateJob } from '@/lib/redis' -import type { Job, JobType, JobCadence, Artifact } from '@/types' - -export async function createJob(params: { - id?: string - userId: string - goal: string - budget: number - type: JobType - cadence?: JobCadence - spendRuleId?: string - qstashScheduleId?: string -}): Promise { - const job: Job = { - id: params.id ?? randomUUID(), - userId: params.userId, - goal: params.goal, - budget: params.budget, - type: params.type, - status: 'pending', - spendRuleId: params.spendRuleId, - spendTotal: 0, - cadence: params.cadence, - qstashScheduleId: params.qstashScheduleId, - createdAt: new Date().toISOString(), - } - - await setJob(job) - return job -} - -export async function startJob(id: string): Promise { - return updateJob(id, { status: 'running' }) -} - -export async function completeJob(id: string, artifact: Artifact): Promise { - return updateJob(id, { - status: 'complete', - artifact, - completedAt: new Date().toISOString(), - }) -} - -export async function failJob(id: string, reason: string): Promise { - return updateJob(id, { - status: 'failed', - failureReason: reason, - completedAt: new Date().toISOString(), - }) -} - -export async function addSpend(id: string, amount: number): Promise { - const job = await getJob(id) - if (!job) return null - return updateJob(id, { spendTotal: job.spendTotal + amount }) -} - -export async function pauseJob(id: string): Promise { - return updateJob(id, { status: 'paused' }) -} - -export async function recordRun( - id: string, - artifact: Artifact, - spentThisRun: number -): Promise { - const job = await getJob(id) - if (!job) return null - - const now = new Date() - const nextRun = job.cadence === 'daily' - ? new Date(now.getTime() + 24 * 60 * 60 * 1000) - : new Date(now.getTime() + 7 * 24 * 60 * 60 * 1000) - - return updateJob(id, { - artifact, - lastRunAt: now.toISOString(), - nextRunAt: nextRun.toISOString(), - spendTotal: (job.spendTotal ?? 0) + spentThisRun, - status: 'running', - }) -} - -export { getJob } +import { randomUUID } from 'crypto' +// Added 'redis' client export to utilize SETNX commands directly +import { setJob, getJob, updateJob, redis } from '@/lib/redis' +import type { Job, JobType, JobCadence, Artifact } from '@/types' + +// SECURITY FIX: Webhook Idempotency locking via SETNX +export async function checkAndLockMessage(messageId: string): Promise { + try { + const key = `webhook:idempotency:${messageId}` + // SETNX returns 1 if the key was set successfully, 0 if it already existed + const acquired = await redis.setnx(key, '1') + if (acquired === 1) { + // Set expiration to clean up old locks (e.g., 7 days) + await redis.expire(key, 86400 * 7) + return false // Not a duplicate + } + return true // Duplicate execution rejected + } catch (err) { + console.warn('Idempotency check failed, defaulting to block:', err) + return true + } +} + +// Releases the lock if the downstream action fails, allowing for a retry +export async function unlockMessage(messageId: string): Promise { + try { + await redis.del(`webhook:idempotency:${messageId}`) + } catch (err) { + console.error('Failed to unlock message:', err) + } +} + +export async function createJob(params: { + id?: string + userId: string + goal: string + budget: number + type: JobType + cadence?: JobCadence + spendRuleId?: string + qstashScheduleId?: string +}): Promise { + const job: Job = { + id: params.id ?? randomUUID(), + userId: params.userId, + goal: params.goal, + budget: params.budget, + type: params.type, + status: 'pending', + spendRuleId: params.spendRuleId, + spendTotal: 0, + cadence: params.cadence, + qstashScheduleId: params.qstashScheduleId, + createdAt: new Date().toISOString(), + } + + await setJob(job) + return job +} + +export async function startJob(id: string): Promise { + const job = await getJob(id) + if (!job) return null + + // SECURITY FIX: Enforce an atomic state transition check. + // Reject duplicate/concurrent executions by ensuring a job can only transition to + // 'running' if it is currently 'pending' or 'paused'. + if (job.status !== 'pending' && job.status !== 'paused') { + throw new Error(`Cannot start job. Current status is '${job.status}'. Duplicate execution rejected.`) + } + + return updateJob(id, { status: 'running' }) +} + +export async function completeJob(id: string, artifact: Artifact): Promise { + return updateJob(id, { + status: 'complete', + artifact, + completedAt: new Date().toISOString(), + }) +} + +export async function failJob(id: string, reason: string): Promise { + return updateJob(id, { + status: 'failed', + failureReason: reason, + completedAt: new Date().toISOString(), + }) +} + +export async function addSpend(id: string, amount: number): Promise { + const job = await getJob(id) + if (!job) return null + return updateJob(id, { spendTotal: job.spendTotal + amount }) +} + +export async function pauseJob(id: string): Promise { + return updateJob(id, { status: 'paused' }) +} + +export async function recordRun( + id: string, + artifact: Artifact, + spentThisRun: number +): Promise { + const job = await getJob(id) + if (!job) return null + + const now = new Date() + const nextRun = job.cadence === 'daily' + ? new Date(now.getTime() + 24 * 60 * 60 * 1000) + : new Date(now.getTime() + 7 * 24 * 60 * 60 * 1000) + + return updateJob(id, { + artifact, + lastRunAt: now.toISOString(), + nextRunAt: nextRun.toISOString(), + spendTotal: (job.spendTotal ?? 0) + spentThisRun, + status: 'running', + }) +} + +export { getJob } \ No newline at end of file