From 16b049919227e4ad4bc894e2b8463f07f4894f02 Mon Sep 17 00:00:00 2001 From: diegosouzapw Date: Wed, 22 Apr 2026 17:10:47 -0300 Subject: [PATCH] fix(api): harden batch and file endpoints for auth and recovery Reject invalid API keys even when authentication is optional and add OPTIONS handlers for batch and file routes to support CORS preflights. Also recover orphaned batches stuck in finalizing after restarts, include finalizing in pending batch queries, preserve multipart upload content handling, and fetch remote vision images as data URIs for Anthropic requests. --- open-sse/services/batchProcessor.ts | 20 ++- open-sse/services/combo.ts | 2 +- src/app/api/v1/_helpers/apiKeyScope.ts | 12 ++ src/app/api/v1/batches/[id]/cancel/route.ts | 6 +- src/app/api/v1/batches/[id]/route.ts | 6 +- src/app/api/v1/batches/route.ts | 6 +- src/app/api/v1/files/[id]/content/route.ts | 6 +- src/app/api/v1/files/[id]/route.ts | 6 +- src/app/api/v1/files/route.ts | 17 +-- src/lib/db/batches.ts | 4 +- src/lib/guardrails/visionBridgeHelpers.ts | 51 ++++++-- tests/unit/batch_api.test.ts | 114 ++++++++++++++++++ ...isionBridgeHelpers.callVisionModel.test.ts | 67 ++++++++-- 13 files changed, 276 insertions(+), 41 deletions(-) diff --git a/open-sse/services/batchProcessor.ts b/open-sse/services/batchProcessor.ts index 15a2a9b455..c0047a5ad0 100644 --- a/open-sse/services/batchProcessor.ts +++ b/open-sse/services/batchProcessor.ts @@ -58,19 +58,27 @@ export function stopBatchProcessor() { } /** - * Mark any in_progress batches as failed on startup. + * Mark any in_progress/finalizing batches as failed on startup. * These were orphaned by a server crash or restart and cannot be safely resumed. */ function recoverOrphanedBatches() { try { const pending = getPendingBatches(); for (const batch of pending) { - if (batch.status === "in_progress") { - console.warn(`[BATCH] Failing orphaned in_progress batch ${batch.id} (server restarted)`); + if (batch.status === "in_progress" || batch.status === "finalizing") { + const interruptedPhase = + batch.status === "finalizing" ? "during finalization" : "while processing requests"; + console.warn( + `[BATCH] Failing orphaned ${batch.status} batch ${batch.id} (server restarted)` + ); updateBatch(batch.id, { status: "failed", failedAt: Math.floor(Date.now() / 1000), - errors: [{ message: "Batch interrupted by server restart and cannot be resumed" }], + errors: [ + { + message: `Batch interrupted ${interruptedPhase} by server restart and cannot be resumed`, + }, + ], }); if (batch.inputFileId) { updateFileStatus(batch.inputFileId, "processed"); @@ -90,8 +98,8 @@ export async function processPendingBatches() { } else if (batch.status === "cancelling") { await cancelBatch(batch); } - // in_progress: currently being processed by processBatchItems running in background; - // orphaned in_progress batches are handled by recoverOrphanedBatches() at startup. + // in_progress/finalizing batches are either actively being worked by the current process + // or will be failed by recoverOrphanedBatches() on the next startup. } // Cleanup task: delete files for batches completed more than completionWindow ago diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 33e845b421..9bcedc2117 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -1982,7 +1982,7 @@ async function handleRoundRobinCombo({ } } - if (isProviderBreakerOpenResponse(result, errorBody as any)) { + if (isProviderBreakerOpenResponse(result, errorBody as Record | null)) { lastError = errorText || String(result.status); if (!lastStatus) lastStatus = result.status; if (offset > 0) fallbackCount++; diff --git a/src/app/api/v1/_helpers/apiKeyScope.ts b/src/app/api/v1/_helpers/apiKeyScope.ts index 07c7afd8ac..cce0f76a75 100644 --- a/src/app/api/v1/_helpers/apiKeyScope.ts +++ b/src/app/api/v1/_helpers/apiKeyScope.ts @@ -40,6 +40,18 @@ export async function getApiKeyRequestScope(request: Request): Promise }) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/batches/[id]/route.ts b/src/app/api/v1/batches/[id]/route.ts index c661fb3a21..e85aa336c4 100644 --- a/src/app/api/v1/batches/[id]/route.ts +++ b/src/app/api/v1/batches/[id]/route.ts @@ -1,4 +1,4 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { getBatch } from "@/lib/localDb"; import { NextResponse } from "next/server"; import { getApiKeyRequestScope } from "@/app/api/v1/_helpers/apiKeyScope"; @@ -34,6 +34,10 @@ function formatBatchResponse(batch: any) { }; } +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function GET(request: Request, { params }: { params: Promise<{ id: string }> }) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/batches/route.ts b/src/app/api/v1/batches/route.ts index f732efb76d..39d24b619d 100644 --- a/src/app/api/v1/batches/route.ts +++ b/src/app/api/v1/batches/route.ts @@ -1,4 +1,4 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { createBatch, getFile, listBatches } from "@/lib/localDb"; import { v1BatchCreateSchema } from "@/shared/validation/schemas"; import { NextResponse } from "next/server"; @@ -35,6 +35,10 @@ function formatBatchResponse(batch: any) { }; } +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function POST(request: Request) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/files/[id]/content/route.ts b/src/app/api/v1/files/[id]/content/route.ts index 4cad037f60..70b1379af7 100644 --- a/src/app/api/v1/files/[id]/content/route.ts +++ b/src/app/api/v1/files/[id]/content/route.ts @@ -1,8 +1,12 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { getFile, getFileContent } from "@/lib/localDb"; import { NextResponse } from "next/server"; import { getApiKeyRequestScope } from "@/app/api/v1/_helpers/apiKeyScope"; +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function GET(request: Request, { params }: { params: Promise<{ id: string }> }) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/files/[id]/route.ts b/src/app/api/v1/files/[id]/route.ts index 3b31db553f..5457349553 100644 --- a/src/app/api/v1/files/[id]/route.ts +++ b/src/app/api/v1/files/[id]/route.ts @@ -1,8 +1,12 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { getFile, deleteFile, formatFileResponse } from "@/lib/localDb"; import { NextResponse } from "next/server"; import { getApiKeyRequestScope } from "@/app/api/v1/_helpers/apiKeyScope"; +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function GET(request: Request, { params }: { params: Promise<{ id: string }> }) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; diff --git a/src/app/api/v1/files/route.ts b/src/app/api/v1/files/route.ts index cceb347f39..16f5b1945b 100644 --- a/src/app/api/v1/files/route.ts +++ b/src/app/api/v1/files/route.ts @@ -1,8 +1,12 @@ -import { CORS_HEADERS } from "@/shared/utils/cors"; +import { CORS_HEADERS, handleCorsOptions } from "@/shared/utils/cors"; import { createFile, listFiles, formatFileResponse } from "@/lib/localDb"; import { NextResponse } from "next/server"; import { getApiKeyRequestScope } from "@/app/api/v1/_helpers/apiKeyScope"; +export async function OPTIONS() { + return handleCorsOptions(); +} + export async function POST(request: Request) { const scope = await getApiKeyRequestScope(request); if (scope.rejection) return scope.rejection; @@ -38,16 +42,7 @@ export async function POST(request: Request) { const bytes = file.size; const filename = file.name; const mimeType = file.type; - - // Stream the upload into memory in chunks to avoid allocating a large contiguous ArrayBuffer - const reader = file.stream().getReader(); - const chunks: Uint8Array[] = []; - while (true) { - const { done, value } = await reader.read(); - if (done) break; - chunks.push(value); - } - const content = Buffer.concat(chunks.map((c) => Buffer.from(c))); + const content = Buffer.from(await file.arrayBuffer()); let expiresAt: number | undefined; if (expiresAfterAnchor === "created_at" && expiresAfterSeconds) { diff --git a/src/lib/db/batches.ts b/src/lib/db/batches.ts index f96dcd99f3..71515f0a7a 100644 --- a/src/lib/db/batches.ts +++ b/src/lib/db/batches.ts @@ -171,7 +171,9 @@ export function listBatches(apiKeyId?: string, limit: number = 20, after?: strin export function getPendingBatches(): BatchRecord[] { const db = getDbInstance(); const rows = db - .prepare("SELECT * FROM batches WHERE status IN ('validating', 'in_progress', 'cancelling')") + .prepare( + "SELECT * FROM batches WHERE status IN ('validating', 'in_progress', 'finalizing', 'cancelling')" + ) .all(); return rows.map((row) => parseBatchRow(row)); } diff --git a/src/lib/guardrails/visionBridgeHelpers.ts b/src/lib/guardrails/visionBridgeHelpers.ts index c0d28376b6..d82cdb334e 100644 --- a/src/lib/guardrails/visionBridgeHelpers.ts +++ b/src/lib/guardrails/visionBridgeHelpers.ts @@ -110,6 +110,34 @@ export function resolveImageAsDataUri(imageUrl: string): string { return `data:image/png;base64,${imageUrl}`; } +async function fetchRemoteImageAsDataUri(imageUrl: string, signal: AbortSignal): Promise { + const response = await fetch(imageUrl, { signal }); + if (!response.ok) { + throw new Error(`Vision image fetch error ${response.status}`); + } + + const mediaType = response.headers.get("content-type")?.split(";")[0]?.trim() || "image/png"; + const bytes = Buffer.from(await response.arrayBuffer()); + return `data:${mediaType};base64,${bytes.toString("base64")}`; +} + +async function normalizeVisionImageInput( + imageInput: string, + isAnthropic: boolean, + signal: AbortSignal +): Promise { + const normalizedImage = resolveImageAsDataUri(imageInput); + + if ( + isAnthropic && + (normalizedImage.startsWith("http://") || normalizedImage.startsWith("https://")) + ) { + return fetchRemoteImageAsDataUri(normalizedImage, signal); + } + + return normalizedImage; +} + export interface VisionModelConfig { model: string; prompt: string; @@ -137,9 +165,12 @@ export async function callVisionModel( try { // Extract model name from provider/model format - const modelName = config.model.includes("/") - ? config.model.split("/")[1] - : config.model; + const modelName = config.model.includes("/") ? config.model.split("/")[1] : config.model; + const normalizedImageInput = await normalizeVisionImageInput( + imageDataUri, + isAnthropic, + controller.signal + ); let response: Response; @@ -148,9 +179,9 @@ export async function callVisionModel( const anthropicBaseUrl = process.env.ANTHROPIC_API_URL || "https://api.anthropic.com"; // Parse data URI to extract media type and base64 data - const matches = imageDataUri.match(/^data:([^;]+);base64,(.+)$/); + const matches = normalizedImageInput.match(/^data:([^;]+);base64,(.+)$/); let mediaType = "image/png"; - let base64Data = imageDataUri; + let base64Data = normalizedImageInput; if (matches) { mediaType = matches[1]; @@ -209,7 +240,7 @@ export async function callVisionModel( { type: "image_url", image_url: { - url: imageDataUri, + url: normalizedImageInput, detail: "low", }, }, @@ -239,7 +270,9 @@ export async function callVisionModel( }; if (anthropicData.error) { - throw new Error(`Vision API error: ${anthropicData.error.message || JSON.stringify(anthropicData.error)}`); + throw new Error( + `Vision API error: ${anthropicData.error.message || JSON.stringify(anthropicData.error)}` + ); } const textContent = anthropicData.content?.find((c) => c.type === "text"); @@ -257,7 +290,9 @@ export async function callVisionModel( }; if (openaiData.error) { - throw new Error(`Vision API error: ${openaiData.error.message || JSON.stringify(openaiData.error)}`); + throw new Error( + `Vision API error: ${openaiData.error.message || JSON.stringify(openaiData.error)}` + ); } const content = openaiData.choices?.[0]?.message?.content; diff --git a/tests/unit/batch_api.test.ts b/tests/unit/batch_api.test.ts index 808dacc5cb..5a4a323e84 100644 --- a/tests/unit/batch_api.test.ts +++ b/tests/unit/batch_api.test.ts @@ -6,6 +6,7 @@ import path from "node:path"; const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-batch-api-")); process.env.DATA_DIR = TEST_DATA_DIR; +process.env.API_KEY_SECRET = process.env.API_KEY_SECRET || "test-secret-123"; const { createFile, @@ -25,7 +26,11 @@ const { getDbInstance } = await import("../../src/lib/db/core.ts"); const { initBatchProcessor, stopBatchProcessor, processPendingBatches } = await import("../../open-sse/services/batchProcessor.ts"); const batchesRoute = await import("../../src/app/api/v1/batches/route.ts"); +const batchByIdRoute = await import("../../src/app/api/v1/batches/[id]/route.ts"); +const batchCancelRoute = await import("../../src/app/api/v1/batches/[id]/cancel/route.ts"); const filesRoute = await import("../../src/app/api/v1/files/route.ts"); +const fileByIdRoute = await import("../../src/app/api/v1/files/[id]/route.ts"); +const fileContentRoute = await import("../../src/app/api/v1/files/[id]/content/route.ts"); test("Batch API and Processing", async () => { // 0. Setup environment, mock provider and API key @@ -654,6 +659,43 @@ test("Batch cleanup honors output_expires_after for output artifacts", async () assert.equal(getFile(errorFile.id), null); }); +test("Batch processor fails orphaned finalizing batches during startup recovery", async () => { + const apiKey = await createApiKey("Finalizing Recovery Key", "test-machine"); + const inputFile = createFile({ + bytes: 2, + filename: "finalizing.jsonl", + purpose: "batch", + content: Buffer.from("{}"), + apiKeyId: apiKey.id, + }); + + const batch = createBatch({ + endpoint: "/v1/chat/completions", + completionWindow: "24h", + inputFileId: inputFile.id, + apiKeyId: apiKey.id, + }); + + updateBatch(batch.id, { + status: "finalizing", + finalizingAt: Math.floor(Date.now() / 1000), + }); + + initBatchProcessor(); + + try { + const recoveredBatch = getBatch(batch.id); + assert.strictEqual(recoveredBatch?.status, "failed"); + assert.match( + String(recoveredBatch?.errors?.[0]?.message || ""), + /interrupted during finalization/i + ); + assert.strictEqual(getFile(inputFile.id)?.status, "processed"); + } finally { + stopBatchProcessor(); + } +}); + test("Batch list route rejects missing API key when REQUIRE_API_KEY is enabled", async () => { const previous = process.env.REQUIRE_API_KEY; process.env.REQUIRE_API_KEY = "true"; @@ -688,6 +730,78 @@ test("Files list route rejects invalid API key when REQUIRE_API_KEY is enabled", } }); +test("Files upload route rejects invalid API key even when auth is optional", async () => { + const previous = process.env.REQUIRE_API_KEY; + process.env.REQUIRE_API_KEY = "false"; + + try { + const formData = new FormData(); + formData.set("purpose", "batch"); + formData.set( + "file", + new File([Buffer.from('{"ok":true}\n')], "input.jsonl", { type: "application/json" }) + ); + + const response = await filesRoute.POST( + new Request("http://localhost/api/v1/files", { + method: "POST", + headers: { Authorization: "Bearer invalid-test-key" }, + body: formData, + }) + ); + const json = await response.json(); + + assert.strictEqual(response.status, 401); + assert.strictEqual(json.error.message, "Invalid API key"); + } finally { + process.env.REQUIRE_API_KEY = previous ?? "false"; + } +}); + +test("Files upload route stores multipart content", async () => { + const fileContent = '{"custom_id":"req-1"}\n'; + const formData = new FormData(); + formData.set("purpose", "batch"); + formData.set( + "file", + new File([Buffer.from(fileContent)], "upload.jsonl", { type: "application/json" }) + ); + + const response = await filesRoute.POST( + new Request("http://localhost/api/v1/files", { + method: "POST", + body: formData, + }) + ); + const json = await response.json(); + + assert.strictEqual(response.status, 200); + assert.ok(json.id); + assert.strictEqual(getFileContent(json.id)?.toString(), fileContent); +}); + +test("Files and batches routes expose explicit CORS preflight handlers", async () => { + const routes = [ + batchesRoute, + batchByIdRoute, + batchCancelRoute, + filesRoute, + fileByIdRoute, + fileContentRoute, + ]; + + for (const route of routes) { + assert.strictEqual(typeof route.OPTIONS, "function"); + const response = await route.OPTIONS(); + assert.strictEqual(response.status, 204); + assert.strictEqual(response.headers.get("Access-Control-Allow-Origin"), "*"); + assert.match( + String(response.headers.get("Access-Control-Allow-Headers") || ""), + /Authorization/i + ); + } +}); + test("Batch Cancel API", async () => { const apiKey = await createApiKey("Cancel Test Key", "test-machine"); diff --git a/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts b/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts index dd241214e1..ed98e35449 100644 --- a/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts +++ b/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts @@ -56,8 +56,7 @@ test("callVisionModel throws on HTTP error", async () => { }; await assert.rejects( - async () => - await callVisionModel("data:image/png;base64,iVBORw0KGgo", config), + async () => await callVisionModel("data:image/png;base64,iVBORw0KGgo", config), /Vision API error 500/ ); } finally { @@ -168,10 +167,7 @@ test("callVisionModel passes custom API key", async () => { await callVisionModel("data:image/png;base64,iVBORw0KGgo", config, "sk-custom-key"); - assert.strictEqual( - capturedHeaders["Authorization"], - "Bearer sk-custom-key" - ); + assert.strictEqual(capturedHeaders["Authorization"], "Bearer sk-custom-key"); } finally { globalThis.fetch = originalFetch; } @@ -210,22 +206,75 @@ test("callVisionModel uses correct request body format", async () => { assert.ok(Array.isArray(capturedBody.messages)); assert.strictEqual((capturedBody.messages as unknown[]).length, 1); - const message = (capturedBody.messages as Array<{role: string; content: unknown[]}>)[0]; + const message = (capturedBody.messages as Array<{ role: string; content: unknown[] }>)[0]; assert.strictEqual(message.role, "user"); assert.ok(Array.isArray(message.content)); assert.strictEqual(message.content.length, 2); // First content is image_url - const imagePart = message.content[0] as {type: string; image_url: {url: string; detail: string}}; + const imagePart = message.content[0] as { + type: string; + image_url: { url: string; detail: string }; + }; assert.strictEqual(imagePart.type, "image_url"); assert.strictEqual(imagePart.image_url.url, imageUri); assert.strictEqual(imagePart.image_url.detail, "low"); // Second content is text prompt - const textPart = message.content[1] as {type: string; text: string}; + const textPart = message.content[1] as { type: string; text: string }; assert.strictEqual(textPart.type, "text"); assert.strictEqual(textPart.text, "What is in this image?"); } finally { globalThis.fetch = originalFetch; } }); + +test("callVisionModel fetches remote images before Anthropic requests", async () => { + const fetchCalls: Array<{ url: string; init?: RequestInit }> = []; + + globalThis.fetch = async (url: URL | RequestInfo, init?: RequestInit) => { + const requestUrl = String(url); + fetchCalls.push({ url: requestUrl, init }); + + if (requestUrl === "https://cdn.example.com/cat.png") { + return new Response(Buffer.from("cat-image-bytes"), { + status: 200, + headers: { "Content-Type": "image/png" }, + }); + } + + return new Response( + JSON.stringify({ + content: [{ type: "text", text: "A cat sitting on a chair" }], + }), + { + status: 200, + headers: { "Content-Type": "application/json" }, + } + ); + }; + + try { + const config: VisionModelConfig = { + model: "anthropic/claude-3-haiku", + prompt: "Describe this image", + timeoutMs: 30000, + maxImages: 10, + }; + + const result = await callVisionModel("https://cdn.example.com/cat.png", config, "sk-ant"); + + assert.strictEqual(result, "A cat sitting on a chair"); + assert.strictEqual(fetchCalls.length, 2); + assert.strictEqual(fetchCalls[0].url, "https://cdn.example.com/cat.png"); + assert.strictEqual(fetchCalls[1].url, "https://api.anthropic.com/v1/messages"); + + const anthropicBody = JSON.parse(fetchCalls[1].init?.body as string); + const imageSource = anthropicBody.messages[0].content[0].source; + assert.strictEqual(imageSource.type, "base64"); + assert.strictEqual(imageSource.media_type, "image/png"); + assert.strictEqual(imageSource.data, Buffer.from("cat-image-bytes").toString("base64")); + } finally { + globalThis.fetch = originalFetch; + } +});