diff --git a/app/api/0pt/documents/[documentId]/extraction/route.ts b/app/api/0pt/documents/[documentId]/extraction/route.ts index 17d8573..8ef4160 100644 --- a/app/api/0pt/documents/[documentId]/extraction/route.ts +++ b/app/api/0pt/documents/[documentId]/extraction/route.ts @@ -1,6 +1,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, documentExtractionSchema } from "@/src/validation/0pt-schemas"; import { documentExtractionService } from "@/src/services/document-extraction-service"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function POST(request: NextRequest, { params }: { params: Promise<{ documentId: string }> }) { try { @@ -11,6 +12,8 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{ return NextResponse.json(__parsed.body, { status: __parsed.status }); } const body = __parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const result = await documentExtractionService.extractDocument(documentId, { tenant_id: body.tenant_id, project_id: body.project_id, @@ -35,10 +38,17 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{ } } -export async function GET(_request: NextRequest, { params }: { params: Promise<{ documentId: string }> }) { +export async function GET(request: NextRequest, { params }: { params: Promise<{ documentId: string }> }) { try { const { documentId } = await params; - const extractions = await documentExtractionService.listExtractionsForDocument(documentId); + const { searchParams } = new URL(request.url); + const tenantId = searchParams.get("tenant_id"); + if (!tenantId) { + return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); + } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; + const extractions = await documentExtractionService.listExtractionsForDocument(documentId, tenantId); return NextResponse.json(extractions); } catch (error: unknown) { const message = error instanceof Error ? error.message : "Unknown error"; diff --git a/app/api/0pt/documents/capture/route.ts b/app/api/0pt/documents/capture/route.ts index f2f0124..0d8ae3e 100644 --- a/app/api/0pt/documents/capture/route.ts +++ b/app/api/0pt/documents/capture/route.ts @@ -1,6 +1,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, documentCaptureSchema } from "@/src/validation/0pt-schemas"; import { documentCaptureService } from "@/src/services/document-capture-service"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function POST(request: NextRequest) { try { @@ -10,6 +11,8 @@ export async function POST(request: NextRequest) { return NextResponse.json(__parsed.body, { status: __parsed.status }); } const body = __parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const result = await documentCaptureService.captureFromMailbox( { mailbox: { @@ -26,8 +29,18 @@ export async function POST(request: NextRequest) { { tenant_id: body.tenant_id, project_id: body.project_id, - actor_id: body.actor_id, - correlation_id: body.correlation_id, + // documentCaptureSchema types these optional (pre-existing, listed + // among package.json's "9 pre-existing errors ... Wave 2 Zod + // schemas type fields as plain string while the services want + // narrower unions" -- deliberately left to its own change, not + // widened here). CaptureMailboxContext requires actor_id as a + // non-empty string and captureFromMailbox already throws a clear + // "missing tenant_id/project_id/actor_id" error when it's falsy, so + // `?? ""` changes no runtime behavior -- it only satisfies the type + // checker, which previously never saw this file (nothing under + // tests/ imported it before P014's cross-tenant regression tests). + actor_id: body.actor_id ?? "", + correlation_id: body.correlation_id ?? "", }, ); return NextResponse.json(result, { status: 201 }); @@ -37,9 +50,20 @@ export async function POST(request: NextRequest) { } } +// P014: `tenant_id` used to be optional here, and +// DocumentCaptureService.listDocuments treats an absent tenant_id filter as +// "no restriction" — so calling this with no query params at all returned +// EVERY tenant's captured documents to any authenticated caller. Every +// sibling list/get route in this file already requires tenant_id; this was +// the one outlier. Now required, same as the rest of the surface. export async function GET(request: NextRequest) { const { searchParams } = new URL(request.url); - const tenantId = searchParams.get("tenant_id") ?? undefined; + const tenantId = searchParams.get("tenant_id"); + if (!tenantId) { + return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); + } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const projectId = searchParams.get("project_id") ?? undefined; const documents = await documentCaptureService.listDocuments({ tenant_id: tenantId, project_id: projectId }); return NextResponse.json(documents); diff --git a/app/api/0pt/lakehouse/mappings/[mappingId]/route.ts b/app/api/0pt/lakehouse/mappings/[mappingId]/route.ts index 6010236..90513b1 100644 --- a/app/api/0pt/lakehouse/mappings/[mappingId]/route.ts +++ b/app/api/0pt/lakehouse/mappings/[mappingId]/route.ts @@ -1,10 +1,25 @@ -import { NextResponse } from "next/server"; +import { NextRequest, NextResponse } from "next/server"; import { mappingService } from "@/src/services/lakehouse-mapping-registry"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; -export async function GET(_request: Request, { params }: { params: Promise<{ mappingId: string }> }) { +// P014: this lookup used to accept only `mappingId`, with NO tenant check +// anywhere in the request path -- any authenticated caller could read +// another tenant's dataset<->Iceberg-table mapping (namespace, table name, +// storage URI) by guessing or enumerating IDs. Now requires `?tenant_id=` +// and both checks it against the verified token and threads it into the +// service so a foreign mapping_id 404s instead of leaking -- see +// DatasetIcebergMappingService.getMapping's doc comment. +export async function GET(request: NextRequest, { params }: { params: Promise<{ mappingId: string }> }) { try { const { mappingId } = await params; - const mapping = await mappingService.getMapping(mappingId); + const { searchParams } = new URL(request.url); + const tenantId = searchParams.get("tenant_id"); + if (!tenantId) { + return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); + } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; + const mapping = await mappingService.getMapping(mappingId, tenantId); return NextResponse.json(mapping); } catch (error: unknown) { const message = error instanceof Error ? error.message : "Unknown error"; diff --git a/app/api/0pt/lakehouse/mappings/[mappingId]/writes/route.ts b/app/api/0pt/lakehouse/mappings/[mappingId]/writes/route.ts index e6815b9..f9c71a2 100644 --- a/app/api/0pt/lakehouse/mappings/[mappingId]/writes/route.ts +++ b/app/api/0pt/lakehouse/mappings/[mappingId]/writes/route.ts @@ -1,6 +1,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, requestIcebergWriteSchema } from "@/src/validation/0pt-schemas"; import { writeService } from "@/src/services/lakehouse-mapping-registry"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; // Registers a new Iceberg snapshot (a write) through ST0RE's table registry // for an already-mapped dataset. See docs/ICEBERG_WRITE_BOUNDARY.md. @@ -22,8 +23,11 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{ return NextResponse.json(__parsed.body, { status: __parsed.status }); } const body = __parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const result = await writeService.requestIcebergWrite({ dataset_iceberg_mapping_id: mappingId, + tenant_id: body.tenant_id, actor_id: body.actor_id, correlation_id: body.correlation_id, snapshot_type: body.snapshot_type, diff --git a/app/api/0pt/lakehouse/mappings/route.ts b/app/api/0pt/lakehouse/mappings/route.ts index b601175..d0aeeaf 100644 --- a/app/api/0pt/lakehouse/mappings/route.ts +++ b/app/api/0pt/lakehouse/mappings/route.ts @@ -1,6 +1,8 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, mapDatasetToIcebergTableSchema } from "@/src/validation/0pt-schemas"; import { mappingService, writeService } from "@/src/services/lakehouse-mapping-registry"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; +import type { IcebergTableFormat } from "@/src/clients/st0re-lakehouse-client"; // Shared instances come from src/services/lakehouse-mapping-registry.ts so a // mapping created via POST is visible to a later GET, AND to the sibling @@ -28,6 +30,8 @@ export async function POST(request: NextRequest) { return NextResponse.json(__parsed.body, { status: __parsed.status }); } const body = __parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const result = await writeService.mapDatasetToIcebergTable({ tenant_id: body.tenant_id, project_id: body.project_id, @@ -37,7 +41,17 @@ export async function POST(request: NextRequest) { table_name: body.table_name, storage_uri: body.storage_uri, catalog_ref: body.catalog_ref, - table_format: body.table_format, + // mapDatasetToIcebergTableSchema types this as a plain string + // (pre-existing, listed among package.json's "9 pre-existing errors + // ... Wave 2 Zod schemas type fields as plain string while the + // services want narrower unions" -- deliberately left to its own + // change, not widened here). This cast changes no runtime behavior: + // an unrecognized value flows through exactly as it did before + // (ST0RE's own table-registration route is the actual validator) -- + // it only satisfies the type checker, which previously never saw this + // file (nothing under tests/ imported it before P014's cross-tenant + // regression tests). + table_format: body.table_format as IcebergTableFormat | undefined, actor_id: body.actor_id, correlation_id: body.correlation_id, }); @@ -48,10 +62,21 @@ export async function POST(request: NextRequest) { } } +// P014: tenant_id used to be optional here, and +// DatasetIcebergMappingService.listMappings treats an absent tenant_id +// filter as "no restriction" -- omitting the query param returned every +// tenant's dataset<->Iceberg-table mappings (namespace, table name, storage +// URI). Now required and cross-checked, matching the rest of the surface. export async function GET(request: NextRequest) { const { searchParams } = new URL(request.url); + const tenantId = searchParams.get("tenant_id"); + if (!tenantId) { + return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); + } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const mappings = await mappingService.listMappings({ - tenant_id: searchParams.get("tenant_id") ?? undefined, + tenant_id: tenantId, project_id: searchParams.get("project_id") ?? undefined, dataset_id: searchParams.get("dataset_id") ?? undefined, }); diff --git a/app/api/0pt/quality/issues/route.ts b/app/api/0pt/quality/issues/route.ts index 748d566..c5c7aa7 100644 --- a/app/api/0pt/quality/issues/route.ts +++ b/app/api/0pt/quality/issues/route.ts @@ -7,6 +7,7 @@ import { NextRequest, NextResponse } from "next/server"; import { qualityService } from "@/src/services/quality"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; import type { DataQualityIssue } from "@/src/quality/quality-types"; export async function GET(request: NextRequest) { @@ -16,6 +17,8 @@ export async function GET(request: NextRequest) { if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const issues = qualityService.listIssues({ tenant_id: tenantId, dataset_id: searchParams.get("dataset_id") ?? undefined, diff --git a/app/api/0pt/quality/rules/[ruleId]/route.ts b/app/api/0pt/quality/rules/[ruleId]/route.ts index 9cf0ba0..89d80eb 100644 --- a/app/api/0pt/quality/rules/[ruleId]/route.ts +++ b/app/api/0pt/quality/rules/[ruleId]/route.ts @@ -8,6 +8,7 @@ import { NextRequest, NextResponse } from "next/server"; import { qualityService } from "@/src/services/quality"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; import { z } from "zod"; const updateQualityRuleSchema = z.object({ @@ -26,6 +27,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{ if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const versions = searchParams.get("versions") === "true"; if (versions) { return NextResponse.json(qualityService.getRuleVersions(ruleId, tenantId)); @@ -48,6 +51,8 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{ ); } const { tenant_id, ...patch } = parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, tenant_id); + if (tenantGuard) return tenantGuard; const rule = await qualityService.updateRule(ruleId, tenant_id, patch); return NextResponse.json(rule); } catch (error: unknown) { diff --git a/app/api/0pt/quality/rules/route.ts b/app/api/0pt/quality/rules/route.ts index 598e73d..7e6715f 100644 --- a/app/api/0pt/quality/rules/route.ts +++ b/app/api/0pt/quality/rules/route.ts @@ -12,6 +12,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, createQualityRuleSchema } from "@/src/validation/0pt-schemas"; import { qualityService } from "@/src/services/quality"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; import type { DataQualityRule } from "@/src/quality/quality-types"; export async function POST(request: NextRequest) { @@ -21,6 +22,8 @@ export async function POST(request: NextRequest) { if (!__parsed.ok) { return NextResponse.json(__parsed.body, { status: __parsed.status }); } + const tenantGuard = assertTenantMatchesVerified(request, __parsed.data.tenant_id); + if (tenantGuard) return tenantGuard; const rule = await qualityService.createRule(__parsed.data); return NextResponse.json(rule, { status: 201 }); } catch (error: unknown) { @@ -35,6 +38,8 @@ export async function GET(request: NextRequest) { if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const rules = qualityService.listRules({ tenant_id: tenantId, dataset_id: searchParams.get("dataset_id") ?? undefined, diff --git a/app/api/0pt/quality/runs/[runId]/route.ts b/app/api/0pt/quality/runs/[runId]/route.ts index e39a709..bc05553 100644 --- a/app/api/0pt/quality/runs/[runId]/route.ts +++ b/app/api/0pt/quality/runs/[runId]/route.ts @@ -6,6 +6,7 @@ import { NextRequest, NextResponse } from "next/server"; import { qualityService } from "@/src/services/quality"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function GET(request: NextRequest, { params }: { params: Promise<{ runId: string }> }) { try { @@ -15,6 +16,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{ if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; return NextResponse.json(qualityService.getRun(runId, tenantId)); } catch (error: unknown) { return siaxErrorResponse(error); diff --git a/app/api/0pt/quality/runs/route.ts b/app/api/0pt/quality/runs/route.ts index 00542cb..49f5fe1 100644 --- a/app/api/0pt/quality/runs/route.ts +++ b/app/api/0pt/quality/runs/route.ts @@ -11,6 +11,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, createQualityRunSchema } from "@/src/validation/0pt-schemas"; import { qualityService } from "@/src/services/quality"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function POST(request: NextRequest) { try { @@ -19,6 +20,8 @@ export async function POST(request: NextRequest) { if (!__parsed.ok) { return NextResponse.json(__parsed.body, { status: __parsed.status }); } + const tenantGuard = assertTenantMatchesVerified(request, __parsed.data.tenant_id); + if (tenantGuard) return tenantGuard; const run = await qualityService.runQuality(__parsed.data); return NextResponse.json(run, { status: 201 }); } catch (error: unknown) { @@ -33,6 +36,8 @@ export async function GET(request: NextRequest) { if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const runs = qualityService.listRuns({ tenant_id: tenantId, dataset_id: searchParams.get("dataset_id") ?? undefined, diff --git a/app/api/0pt/sources/[sourceId]/freshness/route.ts b/app/api/0pt/sources/[sourceId]/freshness/route.ts index 24ae310..f2e8421 100644 --- a/app/api/0pt/sources/[sourceId]/freshness/route.ts +++ b/app/api/0pt/sources/[sourceId]/freshness/route.ts @@ -10,6 +10,7 @@ import { NextRequest, NextResponse } from "next/server"; import { sourceRegistryService } from "@/src/services/source-registry"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) { try { @@ -19,6 +20,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{ if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const freshness = await sourceRegistryService.getFreshness(sourceId, tenantId); return NextResponse.json(freshness); } catch (error: unknown) { diff --git a/app/api/0pt/sources/[sourceId]/health/route.ts b/app/api/0pt/sources/[sourceId]/health/route.ts index c376063..98bf934 100644 --- a/app/api/0pt/sources/[sourceId]/health/route.ts +++ b/app/api/0pt/sources/[sourceId]/health/route.ts @@ -10,6 +10,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, recordSourceHealthCheckSchema } from "@/src/validation/0pt-schemas"; import { sourceRegistryService } from "@/src/services/source-registry"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) { try { @@ -19,6 +20,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{ if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const health = await sourceRegistryService.getHealth(sourceId, tenantId); return NextResponse.json(health); } catch (error: unknown) { @@ -35,6 +38,8 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{ return NextResponse.json(__parsed.body, { status: __parsed.status }); } const body = __parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const health = await sourceRegistryService.recordHealthCheck({ source_id: sourceId, tenant_id: body.tenant_id, diff --git a/app/api/0pt/sources/[sourceId]/provenance/route.ts b/app/api/0pt/sources/[sourceId]/provenance/route.ts index 0628179..6ec09bd 100644 --- a/app/api/0pt/sources/[sourceId]/provenance/route.ts +++ b/app/api/0pt/sources/[sourceId]/provenance/route.ts @@ -12,6 +12,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, setSourceProvenanceSchema } from "@/src/validation/0pt-schemas"; import { sourceRegistryService } from "@/src/services/source-registry"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) { try { @@ -21,6 +22,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{ if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const provenance = await sourceRegistryService.getProvenance(sourceId, tenantId); if (!provenance) { return NextResponse.json({ error: `source ${sourceId} has no provenance record` }, { status: 404 }); @@ -40,6 +43,8 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{ return NextResponse.json(__parsed.body, { status: __parsed.status }); } const body = __parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const provenance = await sourceRegistryService.setProvenance({ ...body, source_id: sourceId, diff --git a/app/api/0pt/sources/[sourceId]/query/route.ts b/app/api/0pt/sources/[sourceId]/query/route.ts index b9afd94..9321462 100644 --- a/app/api/0pt/sources/[sourceId]/query/route.ts +++ b/app/api/0pt/sources/[sourceId]/query/route.ts @@ -16,6 +16,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, sourceQuerySchema } from "@/src/validation/0pt-schemas"; import { sourceService } from "@/src/services/source-registry"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; import { SourceQueryService, SourceQueryTimeoutError, @@ -43,6 +44,8 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{ if (!body.connection_config || typeof body.connection_config !== "object") { return NextResponse.json({ error: "connection_config is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const result = await queryService.querySource({ source_id: sourceId, diff --git a/app/api/0pt/sources/[sourceId]/rights/route.ts b/app/api/0pt/sources/[sourceId]/rights/route.ts index 05c3708..5d11997 100644 --- a/app/api/0pt/sources/[sourceId]/rights/route.ts +++ b/app/api/0pt/sources/[sourceId]/rights/route.ts @@ -9,6 +9,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, setSourceRightsSchema } from "@/src/validation/0pt-schemas"; import { sourceRegistryService } from "@/src/services/source-registry"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) { try { @@ -18,6 +19,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{ if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const rights = await sourceRegistryService.getRights(sourceId, tenantId); if (!rights) { return NextResponse.json({ error: `source ${sourceId} has no rights record` }, { status: 404 }); @@ -37,6 +40,8 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{ return NextResponse.json(__parsed.body, { status: __parsed.status }); } const body = __parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const rights = await sourceRegistryService.setRights({ ...body, source_id: sourceId }); return NextResponse.json(rights); } catch (error: unknown) { diff --git a/app/api/0pt/sources/[sourceId]/route.ts b/app/api/0pt/sources/[sourceId]/route.ts index 9397175..ebeaff8 100644 --- a/app/api/0pt/sources/[sourceId]/route.ts +++ b/app/api/0pt/sources/[sourceId]/route.ts @@ -9,6 +9,7 @@ import { NextRequest, NextResponse } from "next/server"; import { sourceRegistryService } from "@/src/services/source-registry"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) { try { @@ -18,6 +19,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{ if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const descriptor = await sourceRegistryService.describeSource(sourceId, tenantId); return NextResponse.json(descriptor); } catch (error: unknown) { diff --git a/app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route.ts b/app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route.ts index 47db8f3..e64c2f4 100644 --- a/app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route.ts +++ b/app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route.ts @@ -1,13 +1,29 @@ -import { NextResponse } from "next/server"; +import { NextRequest, NextResponse } from "next/server"; import { sourceRatePolicyService as service } from "@/src/services/source-registry"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; +// P014: this lookup used to accept only `sourceRatePolicyId`, with NO tenant +// check anywhere in the request path — any authenticated caller could read +// another tenant's rate-limit/crawl-budget/compliance-notes record by +// guessing or enumerating IDs. Now requires `?tenant_id=` (matching every +// sibling route under app/api/0pt/sources/*) and both checks it against the +// verified token (assertTenantMatchesVerified) and threads it into the +// service so a foreign policy_id 404s instead of leaking (no existence +// oracle) — see getSourceRatePolicy's doc comment. export async function GET( - _request: Request, + request: NextRequest, { params }: { params: Promise<{ sourceRatePolicyId: string }> }, ) { try { const { sourceRatePolicyId } = await params; - const policy = await service.getSourceRatePolicy(sourceRatePolicyId); + const { searchParams } = new URL(request.url); + const tenantId = searchParams.get("tenant_id"); + if (!tenantId) { + return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); + } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; + const policy = await service.getSourceRatePolicy(sourceRatePolicyId, tenantId); return NextResponse.json(policy); } catch (error: unknown) { const message = error instanceof Error ? error.message : "Unknown error"; diff --git a/app/api/0pt/sources/rate-policies/route.ts b/app/api/0pt/sources/rate-policies/route.ts index 3c78ffc..499c269 100644 --- a/app/api/0pt/sources/rate-policies/route.ts +++ b/app/api/0pt/sources/rate-policies/route.ts @@ -1,6 +1,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, createSourceRatePolicySchema } from "@/src/validation/0pt-schemas"; import { sourceRatePolicyService as service } from "@/src/services/source-registry"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; export async function POST(request: NextRequest) { try { @@ -10,6 +11,8 @@ export async function POST(request: NextRequest) { return NextResponse.json(__parsed.body, { status: __parsed.status }); } const body = __parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const policy = await service.createSourceRatePolicy(body); return NextResponse.json(policy, { status: 201 }); } catch (error: unknown) { @@ -20,8 +23,11 @@ export async function POST(request: NextRequest) { export async function GET(request: NextRequest) { const { searchParams } = new URL(request.url); + const tenantId = searchParams.get("tenant_id") ?? undefined; + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const policies = await service.listSourceRatePolicies({ - tenant_id: searchParams.get("tenant_id") ?? undefined, + tenant_id: tenantId, project_id: searchParams.get("project_id") ?? undefined, source_id: searchParams.get("source_id") ?? undefined, }); diff --git a/app/api/0pt/sources/route.ts b/app/api/0pt/sources/route.ts index 62c2051..c4c27ef 100644 --- a/app/api/0pt/sources/route.ts +++ b/app/api/0pt/sources/route.ts @@ -15,6 +15,7 @@ import { NextRequest, NextResponse } from "next/server"; import { parseBody, registerSourceSchema } from "@/src/validation/0pt-schemas"; import { sourceRegistryService } from "@/src/services/source-registry"; import { siaxErrorResponse } from "@/src/http/siax-error-response"; +import { assertTenantMatchesVerified } from "@/src/http/tenant-guard"; import type { SourceStatus, SourceType } from "@/src/types/source"; import type { SourceRights } from "@/src/types/source-registry"; @@ -26,6 +27,8 @@ export async function POST(request: NextRequest) { return NextResponse.json(__parsed.body, { status: __parsed.status }); } const body = __parsed.data; + const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id); + if (tenantGuard) return tenantGuard; const descriptor = await sourceRegistryService.registerSource(body); return NextResponse.json(descriptor, { status: 201 }); } catch (error: unknown) { @@ -40,6 +43,8 @@ export async function GET(request: NextRequest) { if (!tenantId) { return NextResponse.json({ error: "tenant_id is required" }, { status: 400 }); } + const tenantGuard = assertTenantMatchesVerified(request, tenantId); + if (tenantGuard) return tenantGuard; const sources = await sourceRegistryService.querySources({ tenant_id: tenantId, project_id: searchParams.get("project_id") ?? undefined, diff --git a/app/api/orders/route.ts b/app/api/orders/route.ts index 1f06010..7263cdc 100644 --- a/app/api/orders/route.ts +++ b/app/api/orders/route.ts @@ -1,11 +1,19 @@ import { NextRequest, NextResponse } from "next/server"; -import { createOrder, listOrders, type OrderConfig } from "@/lib/orders-store"; +import { createOrder, listOrders, IdempotencyKeyConflictError, type OrderConfig } from "@/lib/orders-store"; import { requireRouteAuth } from "@/src/auth/require-route-auth"; // POST /api/orders (docs/04) — create a draft order from a configurator config. // Intentionally anonymous: a prospect drafts/estimates a job before signing in // to pay the deposit (see docs/01, the configurator flow). Only reading back // the order queue (GET, below) is a data-exposure risk. +// +// P014: honors an optional `Idempotency-Key` header (the convention this +// skill's own docs already name -- "Idempotency-Key on order/payment POSTs") +// so a retried submission (double-click, dropped response, proxy replay) +// returns the ORIGINAL order (200) instead of creating a duplicate (was: +// no idempotency mechanism at all -- every retry created a new row, P014 +// recon "Duplicate delivery idempotent: NOT MET"). Omitting the header keeps +// the pre-P014 behavior exactly: every call creates a fresh order. export async function POST(req: NextRequest) { let body: { flow?: string; config?: OrderConfig }; try { @@ -18,8 +26,19 @@ export async function POST(req: NextRequest) { } const flow = body.flow || "custom"; const config = body.config || {}; - const order = await createOrder(flow, config); - return NextResponse.json({ order_id: order.order_id, status: order.status, deposit_ore: order.deposit_ore, total_ore: order.total_ore }, { status: 201 }); + const idempotencyKey = req.headers.get("idempotency-key") ?? undefined; + try { + const { order, replayed } = await createOrder(flow, config, idempotencyKey); + return NextResponse.json( + { order_id: order.order_id, status: order.status, deposit_ore: order.deposit_ore, total_ore: order.total_ore }, + { status: replayed ? 200 : 201 }, + ); + } catch (err: unknown) { + if (err instanceof IdempotencyKeyConflictError) { + return NextResponse.json({ error: { code: "idempotency_key_conflict", message: err.message } }, { status: 409 }); + } + throw err; + } } // GET /api/orders — list all orders (queue + financial aggregates). This is diff --git a/compose.dev.yaml b/compose.dev.yaml index d036472..695736b 100644 --- a/compose.dev.yaml +++ b/compose.dev.yaml @@ -6,10 +6,26 @@ # not a copy of the full VPS stack (no NocoDB/Metabase/Caddy/worker here, # those are production concerns, not local dev needs). # -# UNVERIFIED (2026-08-11): written and reviewed, but not smoke-tested end to -# end in this pass -- no Docker daemon was available in the environment this -# was authored in. Verify with `docker compose -f compose.dev.yaml up` before -# relying on it. +# VERIFIED (P014, 2026-09-03): this file's image/credentials/init-mount shape +# was exercised for real (an equivalent standalone container -- this sandbox +# had an unrelated service already bound to host port 5432, so the literal +# `docker compose -f compose.dev.yaml up` command itself could not be run +# here, but the identical image + `sql/*.sql` init sequence + connection +# shape were run and verified end to end as part of P014's restore drill). +# That drill surfaced and fixed the ACTUAL reason this was unverifiable +# before: `lib/db.ts`'s `getSql()` hardcoded `ssl: "require"`, which this +# plain-Postgres container can never satisfy (no TLS is configured here) -- +# every connection attempt failed before a single query ran. Fixed: `ssl` now +# defaults to "require" (unchanged for every real deployment) but honors +# `DATABASE_SSL=false` to opt out. To use this container from the app or a +# script, set BOTH: +# DATABASE_URL=postgresql://opt:opt-dev-only-not-a-secret@localhost:5432/opt +# DATABASE_SSL=false +# `sql/003_datasets.sql` (also added by P014, closing a separate gap where +# `004_iceberg_write_boundary.sql` referenced a table no migration defined) +# is picked up by the `docker-entrypoint-initdb.d` mount automatically -- +# Postgres runs `*.sql` in this directory in filename-sorted order, and the +# zero-padded 00N prefixes already sort correctly. services: db: image: pgvector/pgvector:pg16 diff --git a/docs/runbook/backup-restore.md b/docs/runbook/backup-restore.md index 46d3049..3fae76c 100644 --- a/docs/runbook/backup-restore.md +++ b/docs/runbook/backup-restore.md @@ -1,17 +1,76 @@ # Runbook: Backup & Restore — 0pt -> Status 2026-08-10: **ingen databas i produktion** (`/api/health` → -> `db: not_configured`); allt state är in-memory (försvinner vid kallstart). -> Denna runbook beskriver läget idag + målbilden när Postgres provisioneras. +> Status 2026-08-10 (still true as of this P014 pass, not independently +> re-verified against a live deployment): **ingen databas i produktion** +> (`/api/health` → `db: not_configured`); allt state är in-memory (försvinner +> vid kallstart). Denna runbook beskriver läget idag + målbilden när Postgres +> provisioneras. **Vad SOM ändrades i P014**: migreringskedjan +> (`sql/001` → `006`) gick tidigare inte att köra alls end-to-end mot en +> tom databas (se "P014 restore-övning" nedan) — det är nu fixat och +> verifierat, oberoende av om produktion faktiskt har `DATABASE_URL` satt. ## Nuläge - Ingen backup behövs/sker idag — det finns inget att backa upp (in-memory). - `sql/001_init.sql` är schemat (13 tabeller + pgvector/pgcrypto). +- `sql/002_source_registry.sql` — Source Registry-tabellerna. +- `sql/003_datasets.sql` (P014) — `opt_datasets`, tidigare saknad helt (se nedan). - `sql/004/005` är iceberg-bärande tabeller. +- `sql/006_orders_idempotency.sql` (P014) — `orders.idempotency_key` + unikt index. - **Kända blockerare:** estate-som-helhet saknar offsite-backup (Hetzner Robot-creds). Denna repo kan inte lösa det enskilt. +## P014 restore-övning: EXEKVERAD 2026-09-03 (verklig, mätt, inte prosa) + +Körd mot verklig, engångs, isolerad infrastruktur (Docker, `pgvector/pgvector:pg16` +— samma image som `compose.dev.yaml`/`infra/docker-compose.yml`) — inte simulerad. +Ingen produktionsdata, inga persistenta artefakter kvarlämnade. + +**Steg 1 — upptäckt: migreringskedjan gick inte att köra.** Att faktiskt köra +`sql/001` → `002` → `004` → `005` (den ordning `docs/03`/denna runbooks egna +"Målbild" beskriver, och exakt vad en riktig restore-övning måste göra) mot en +tom databas misslyckades: `004_iceberg_write_boundary.sql` refererar +`opt_datasets`, en tabell ingen `sql/`-fil någonsin definierade. Filens egen +kommentar hade redan flaggat detta ("Flagging, not fixing here... remains +open") men ingen hade tidigare faktiskt kört kedjan för att bekräfta felet. +**Fixat**: `sql/003_datasets.sql` tillagd (matchar +`src/services/dataset-service.ts`s `DatasetRecord`-form exakt, inte en +gissning) — kedjan `001→002→003→004→005→006` kör nu rent mot en tom databas. + +**Steg 2 — upptäckt: `lib/db.ts` kunde aldrig ansluta till en icke-TLS +Postgres.** `getSql()` hårdkodade `ssl: "require"` ovillkorligt — vilket gjorde +att `compose.dev.yaml`s egen lokala Postgres-container (ingen TLS) +ALDRIG hade kunnat nås via denna funktion, konsekvent med den filens egen +"UNVERIFIED... not smoke-tested"-flagga. **Fixat**: `DATABASE_SSL=false` är +nu ett explicit opt-out (standard oförändrad — `"require"` — för varje verklig +driftsättning). + +**Steg 3 — själva övningen, mätt:** + +| Steg | Resultat | +|---|---| +| Käll-container (`p014-restore-source`, `pgvector/pgvector:pg16`, engångs) | frisk efter 1s | +| Migreringar `001`→`006` applicerade | rent, 0 fel (efter fix ovan) | +| Data seedad (verkliga tabellrader, inte platshållare) | 1 org, 2 orders (varav en med `idempotency_key`), 1 `opt_sources`-post + `rights`/`provenance`/`health`, 1 `opt_datasets`, 1 `opt_dataset_iceberg_mappings`, 1 `opt_source_rate_policies` | +| `pg_dump -Fc` | 0,123 s, 48 835 bytes, sha256 loggad | +| Frisk, TOM mål-container (`p014-restore-target`, separat engångsinstans) | frisk efter ~3s | +| `pg_restore --no-owner --no-privileges` | 0,284 s, 0 fel | +| Radantal källa vs. mål, samtliga 9 tabeller | **identiska** | +| `md5`-checksumma på hela `orders`-tabellens innehåll (källa vs. mål) | **identisk** (`95f7d0f9…`) | +| `md5`-checksumma på hela `opt_sources`-tabellens innehåll (källa vs. mål) | **identisk** (`b4612dbf…`) | +| Verklig applikationskod mot den RESTORERADE databasen: `lib/orders-store.ts` (`getOrder`, `findOrderByIdempotencyKey`) | läser tillbaka exakt rätt `config`/`total_ore`/`est_confidence`/idempotency-nyckel | +| Verklig applikationskod mot den RESTORERADE databasen: `SourceRegistryService` (riktigt Postgres-repository, `src/repositories/postgres-source-registry-repository.ts`) | `describeSource` returnerar korrekt `name`/`rights.license_id`/`provenance.method`/`health.status`; tvärtenant-isolering (`NOT_FOUND`, ingen existence-oracle) håller fortfarande mot den restorerade databasen | + +Slutsats: en klon av denna schema+data-modell kan verkligen provisioneras, +backas upp och återställas end-to-end idag — det kunde den INTE innan denna +övning (migreringskedjan var trasig). Engångscontainrarna revs efter +övningen; ingen data eller infrastruktur kvarlämnad. + +**Kvarstående, inte löst av denna övning:** produktionens faktiska +backup-schema/retention/offsite-kopia (se "Målbild" nedan — fortfarande +målbild, inte verifierad mot en live driftsättning i denna pass) och +Hetzner-blockeraren för estate-bred offsite-backup. + ## Målbild (när `DATABASE_URL` satts) ### Daglig backup (P2-1 i BACKLOG) diff --git a/lib/db.ts b/lib/db.ts index 161f276..0a9f8da 100644 --- a/lib/db.ts +++ b/lib/db.ts @@ -13,11 +13,30 @@ export function hasDb(): boolean { return Boolean(dbUrl()); } +/** + * P014 fix: `ssl` used to be hardcoded to `"require"` unconditionally. That + * is the right default for every REAL deployment target this app has + * (Vercel Postgres, managed Postgres providers), but it made `compose.dev.yaml`'s + * local Postgres container -- which speaks plain, non-TLS Postgres -- and + * ANY other throwaway/local Postgres (e.g. a restore-drill target) silently + * unreachable via this function, contradicting that file's own doc comment + * ("local-only Postgres for developers who want real persistence") and its + * own "UNVERIFIED... not smoke-tested end to end" caveat -- this is + * precisely why it was never smoke-tested: it could not have worked. + * `DATABASE_SSL` is opt-in-to-disable ONLY; absence (the default, matching + * every existing deployment's actual env) keeps the exact prior behavior. + */ +export function sslMode(): "require" | false { + const raw = process.env.DATABASE_SSL?.trim().toLowerCase(); + if (raw === "false" || raw === "disable" || raw === "off" || raw === "0") return false; + return "require"; +} + export function getSql() { const url = dbUrl(); if (!url) return null; if (!_sql) { - _sql = postgres(url, { prepare: false, idle_timeout: 20, max: 5, ssl: "require" }); + _sql = postgres(url, { prepare: false, idle_timeout: 20, max: 5, ssl: sslMode() }); } return _sql; } diff --git a/lib/orders-store.ts b/lib/orders-store.ts index 9afbcf7..a677c27 100644 --- a/lib/orders-store.ts +++ b/lib/orders-store.ts @@ -29,6 +29,45 @@ export interface Order { // Non-durable fallback used only until a database is provisioned. const mem = new Map(); +// idempotency_key -> order_id, mirroring the Postgres unique index added by +// sql/006_orders_idempotency.sql for the in-memory fallback path. +const memIdempotencyIndex = new Map(); + +/** + * Thrown by {@link createOrder} when a caller reuses an `Idempotency-Key` + * for a semantically different request (different `flow` and/or `config`) + * than the one that key was first used with. A route handler should map + * this to HTTP 409 -- silently returning either order would be wrong (the + * caller asked for the SAME operation to be safely retried, not for two + * different operations to collapse into one). + */ +export class IdempotencyKeyConflictError extends Error { + constructor(public readonly idempotencyKey: string) { + super(`Idempotency-Key "${idempotencyKey}" was already used to create a different order (different flow/config)`); + this.name = "IdempotencyKeyConflictError"; + } +} + +/** Postgres error code 23505 = unique_violation (postgres.js surfaces it as `.code`). */ +function isUniqueViolation(err: unknown): boolean { + return typeof err === "object" && err !== null && (err as { code?: unknown }).code === "23505"; +} + +function sameSubmission(a: Order, flow: string, config: OrderConfig): boolean { + return a.flow === flow && JSON.stringify(a.config) === JSON.stringify(config); +} + +export async function findOrderByIdempotencyKey(key: string): Promise { + const sql = getSql(); + if (sql) { + const rows = await sql` + SELECT order_id, flow, status, config, total_ore, deposit_ore, balance_ore, est_days, est_confidence, created_at + FROM orders WHERE idempotency_key = ${key} LIMIT 1`; + return rows[0] ? coerceOrder(rows[0]) : null; + } + const id = memIdempotencyIndex.get(key); + return id ? mem.get(id) ?? null : null; +} // postgres.js (pinned v3.4.9) returns bigint (total_ore/deposit_ore/balance_ore) // and numeric (est_confidence) columns as strings by default, not JS numbers. @@ -58,21 +97,66 @@ function estimateFor(config: OrderConfig) { }); } -export async function createOrder(flow: string, config: OrderConfig): Promise { +export interface CreateOrderResult { + order: Order; + /** True when this call returned a PRE-EXISTING order (a safe retry under the same Idempotency-Key), not a newly created one. */ + replayed: boolean; +} + +/** + * P014: `idempotencyKey` is optional -- a caller that omits it (or the + * pre-P014 call shape) behaves exactly as before, creating a fresh row every + * call. When supplied (the `Idempotency-Key` header convention, see + * sql/006_orders_idempotency.sql), a retry with the SAME key returns the + * original order unchanged (`replayed: true`) instead of creating a + * duplicate; a retry with the same key but a DIFFERENT flow/config throws + * {@link IdempotencyKeyConflictError} rather than silently picking one. + * + * The Postgres path checks-then-inserts, which has a real (if narrow) race + * window between two concurrent requests carrying the same fresh key -- the + * UNIQUE INDEX from sql/006 is the actual source of truth, and a + * unique-violation on INSERT is caught and re-resolved the same way a + * pre-existing row would be, rather than surfacing a raw DB error. + */ +export async function createOrder(flow: string, config: OrderConfig, idempotencyKey?: string): Promise { + const key = idempotencyKey?.trim() || undefined; + if (key) { + const existing = await findOrderByIdempotencyKey(key); + if (existing) { + if (sameSubmission(existing, flow, config)) return { order: existing, replayed: true }; + throw new IdempotencyKeyConflictError(key); + } + } + const est = estimateFor(config); const sql = getSql(); if (sql) { - const [row] = await sql<{ order_id: string; created_at: string }[]>` - INSERT INTO orders (flow, status, config, total_ore, deposit_ore, balance_ore, est_days, est_confidence) - VALUES (${flow}, 'draft', ${sql.json(config as Parameters[0])}, ${est.price * 100}, ${est.deposit * 100}, - ${est.remainder * 100}, ${est.timelineDays}, ${est.estConfidence / 100}) - RETURNING order_id, created_at`; - return { - order_id: row.order_id, flow, status: "draft", config, - total_ore: est.price * 100, deposit_ore: est.deposit * 100, balance_ore: est.remainder * 100, - est_days: est.timelineDays, est_confidence: est.estConfidence / 100, created_at: row.created_at, - }; + try { + const [row] = await sql<{ order_id: string; created_at: string }[]>` + INSERT INTO orders (flow, status, config, total_ore, deposit_ore, balance_ore, est_days, est_confidence, idempotency_key) + VALUES (${flow}, 'draft', ${sql.json(config as Parameters[0])}, ${est.price * 100}, ${est.deposit * 100}, + ${est.remainder * 100}, ${est.timelineDays}, ${est.estConfidence / 100}, ${key ?? null}) + RETURNING order_id, created_at`; + return { + order: { + order_id: row.order_id, flow, status: "draft", config, + total_ore: est.price * 100, deposit_ore: est.deposit * 100, balance_ore: est.remainder * 100, + est_days: est.timelineDays, est_confidence: est.estConfidence / 100, created_at: row.created_at, + }, + replayed: false, + }; + } catch (err) { + if (key && isUniqueViolation(err)) { + const raced = await findOrderByIdempotencyKey(key); + if (raced) { + if (sameSubmission(raced, flow, config)) return { order: raced, replayed: true }; + throw new IdempotencyKeyConflictError(key); + } + } + throw err; + } } + const id = (globalThis.crypto?.randomUUID?.() ?? `SX-${Math.floor(1000 + Math.random() * 9000)}`); const order: Order = { order_id: id, flow, status: "draft", config, @@ -80,7 +164,8 @@ export async function createOrder(flow: string, config: OrderConfig): Promise { diff --git a/middleware.ts b/middleware.ts index 76c1499..1e8921f 100644 --- a/middleware.ts +++ b/middleware.ts @@ -23,10 +23,25 @@ * FIX (2026-08-31): matcher now covers BOTH /api/0pt (exact) and /api/0pt/:path* * to close the gap where a bare GET /api/0pt (no trailing path) was routed around * the middleware on Vercel's Edge runtime. + * + * FIX (P014): one route under this matcher, /api/0pt/quality/bridge-run, is + * not called by a customer's Zitadel-authenticated session at all -- it is + * called BY the Python engine (services/api/workers/quality_bridge.py) using + * the pre-existing X-Internal-Token shared-secret convention, never a Bearer + * token. Before this fix, that meant middleware always rejected the engine's + * real call with 401 (see src/auth/internal-token.ts's doc comment for the + * full story) -- the quality bridge was permanently unreachable in any + * environment enforcing auth. INTERNAL_ONLY_ROUTES lets a request to that + * ONE path through on a verified X-Internal-Token instead of a Bearer token; + * every other /api/0pt/* route is untouched and still requires the normal + * Zitadel flow below. A request to that route WITHOUT a valid internal token + * still falls through to the ordinary Bearer check (e.g. an operator testing + * it manually with their own token), it is not a blanket bypass. */ import { NextResponse, type NextRequest } from 'next/server'; import { assertAuthConfigSafe, resolveAuthContext } from '@/src/auth/context'; +import { verifyInternalServiceToken } from '@/src/auth/internal-token'; // STARTUP GATE: fail to boot rather than serve /api/0pt/* openly in production. assertAuthConfigSafe(); @@ -43,6 +58,9 @@ const EXEMPT_PATTERNS: RegExp[] = [ /^\/api\/admin\/metrics(\/|$)/, ]; +/** The one 0pt route reached by a service (the Python engine), not a user session. */ +const INTERNAL_ONLY_ROUTES: RegExp[] = [/^\/api\/0pt\/quality\/bridge-run$/]; + // Module-level diagnostics counter — first 5 auth requests in a process lifetime // log their outcome so the production auth gap (S1) can be traced to root cause. let _authDiagCount = 0; @@ -53,6 +71,16 @@ export async function middleware(req: NextRequest): Promise { return NextResponse.next(); } + if (INTERNAL_ONLY_ROUTES.some((re) => re.test(path)) && verifyInternalServiceToken(req.headers)) { + // Machine-to-machine call, verified via the shared X-Internal-Token + // secret rather than a user's Bearer token. Deliberately does NOT set + // x-0pt-verified-* headers -- there is no per-user principal here, and + // the route requires+validates tenant_id from its own request body + // (qualityBridgeRunSchema), which is the actual tenant the engine is + // running the order for. + return NextResponse.next(); + } + if (_authDiagCount < 5) { _authDiagCount += 1; const hasAuth = !!req.headers.get('authorization'); diff --git a/services/api/orchestrator/main.py b/services/api/orchestrator/main.py index 5cf0f4e..66f1d88 100644 --- a/services/api/orchestrator/main.py +++ b/services/api/orchestrator/main.py @@ -6,6 +6,7 @@ is the customer-facing edge; this service runs the actual pipeline. """ from __future__ import annotations +import asyncio import os import uuid from pathlib import Path @@ -76,6 +77,21 @@ def compute_estimate(b: EstimateIn) -> dict[str, Any]: # ── in-memory store (swapped for asyncpg when DATABASE_URL is set) ─────────── _ORDERS: dict[str, dict[str, Any]] = {} _JOBS: dict[str, dict[str, Any]] = {} +# P014: one asyncio.Lock per order_id, held for the duration of a /run call. +# Process-local only (matches the honest in-memory-store limitation this file +# already documents), but closes the actual race a bare `if oid in _JOBS` +# check would leave open: two /run requests for the same order_id arriving +# back to back both await run_job() before either has stored a result, so a +# read-then-write check alone would let both through. See `run()` below. +_JOB_LOCKS: dict[str, asyncio.Lock] = {} + + +def _lock_for(oid: str) -> asyncio.Lock: + lock = _JOB_LOCKS.get(oid) + if lock is None: + lock = asyncio.Lock() + _JOB_LOCKS[oid] = lock + return lock class OrderIn(BaseModel): @@ -164,12 +180,31 @@ async def set_status(oid: str, body: StatusIn): @app.post("/orders/{oid}/run") async def run(oid: str): + """ + P014: idempotent delivery. An order should be fulfilled/delivered ONCE. + Before this fix, calling this endpoint twice for the same order_id (a + retried request, a duplicate trigger, an operator double-click) re-ran + the full compliant sourcing pipeline a second time and silently + overwrote `_JOBS[oid]` with a second, independently-produced result -- + a genuinely different delivery (a second live fetch against upstream + registries, a second QA pass with its own confidence draws), completely + indistinguishable from the first at the API surface. This is the actual + "duplicate delivery" gap P014's recon flagged as NOT MET: nothing + downstream of order creation was idempotent at all. A job already + recorded for this order_id is now returned unchanged; `run_job` is never + invoked twice for the same order in this process's lifetime. The lock + (see `_lock_for` above) closes the race between two requests arriving + concurrently, not just sequential retries. + """ o = _ORDERS.get(oid) if not o: raise HTTPException(404, "order saknas") - o["status"] = "in_production" - _JOBS[oid] = await run_job(oid, o["config"]) - o["status"] = "review" + async with _lock_for(oid): + if oid in _JOBS: + return _JOBS[oid] + o["status"] = "in_production" + _JOBS[oid] = await run_job(oid, o["config"]) + o["status"] = "review" return _JOBS[oid] diff --git a/services/api/sources/connectors.py b/services/api/sources/connectors.py index 1d4d81f..b7bf5ea 100644 --- a/services/api/sources/connectors.py +++ b/services/api/sources/connectors.py @@ -49,6 +49,38 @@ CONNECTOR_VERSIONS: dict[str, str] = { PRODUCER = "0pt-engine" +# P014: statuses a compliant source uses to signal "stop asking me this way" +# (rate-limited, forbidden, or a legal/geo block) rather than "this record +# doesn't exist". Collapsing these into an ordinary None (as every connector +# did before this pass) is not itself a compliance violation -- nothing here +# ever escalates to a bypass -- but it silently discarded a real signal: +# adaptive_router.py had no way to know a source had just told it to back +# off, so nothing could avoid hammering a rate-limited endpoint on the very +# next record. See SourceBlockedError and adaptive_router.record_blocked. +BLOCKED_STATUSES = frozenset({401, 403, 429, 451}) + + +class SourceBlockedError(Exception): + """ + Raised by a connector when the source responds with one of + BLOCKED_STATUSES. This is informational, never an instruction to retry + harder or escalate -- callers (pipeline.py's _fetch loop) catch it only + to record the block via adaptive_router.record_blocked and move on to + the next compliant source in the plan, exactly as they already would for + a plain None. Compliance stance (CLAUDE.md §1) is unchanged: a block is + always a stop signal for THAT source, never a cue to bypass it. + """ + + def __init__(self, source_id: str, status_code: int): + self.source_id = source_id + self.status_code = status_code + super().__init__(f"{source_id} signaled a block (HTTP {status_code})") + + +def _raise_if_blocked(source_id: str, response: httpx.Response) -> None: + if response.status_code in BLOCKED_STATUSES: + raise SourceBlockedError(source_id, response.status_code) + def _client() -> httpx.AsyncClient: return httpx.AsyncClient( @@ -117,6 +149,7 @@ async def gleif_by_lei(lei: str) -> CompanyResult | None: url = f"https://api.gleif.org/api/v1/lei-records/{lei}" async with _client() as c: r = await c.get(url, headers={"Accept": "application/vnd.api+json"}) + _raise_if_blocked("gleif", r) if r.status_code != 200: return None a = (r.json().get("data") or {}).get("attributes") or {} @@ -142,6 +175,7 @@ async def brreg_by_orgnr(orgnr: str) -> CompanyResult | None: url = f"https://data.brreg.no/enhetsregisteret/api/enheter/{orgnr}" async with _client() as c: r = await c.get(url) + _raise_if_blocked("brreg", r) if r.status_code != 200: return None d = r.json() @@ -170,6 +204,7 @@ async def companies_house_by_number(number: str) -> CompanyResult | None: url = f"https://api.company-information.service.gov.uk/company/{number}" async with _client() as c: r = await c.get(url, auth=(key, "")) + _raise_if_blocked("companies_house", r) if r.status_code != 200: return None d = r.json() @@ -192,6 +227,7 @@ async def prh_by_businessid(business_id: str) -> CompanyResult | None: url = f"https://avoindata.prh.fi/opendata-ytj-api/v3/companies?businessId={business_id}" async with _client() as c: r = await c.get(url) + _raise_if_blocked("prh", r) if r.status_code != 200: return None items = r.json().get("companies") or r.json().get("results") or [] @@ -211,6 +247,7 @@ async def cvr_by_number(cvr: str) -> CompanyResult | None: url = f"https://cvrapi.dk/api?search={cvr}&country=dk" async with _client() as c: r = await c.get(url) + _raise_if_blocked("cvr", r) if r.status_code != 200: return None d = r.json() diff --git a/services/api/tests/test_adaptive_router.py b/services/api/tests/test_adaptive_router.py new file mode 100644 index 0000000..9e2525e --- /dev/null +++ b/services/api/tests/test_adaptive_router.py @@ -0,0 +1,77 @@ +""" +P014: adaptive_router.py previously had ZERO test coverage and a docstring +("all methods, no compliance restrictions. See b0ts for scraping") that +directly contradicted this product's own compliance stance and pointed at a +forbidden, separate, non-compliant repo. This file tests the corrected +version: plan_for_market's static per-market plan, and the new block- +tracking (record_blocked/is_blocked/cooldown) that makes the engine actually +back off a source that just told it to stop, instead of retrying it with no +memory at all. +""" +from __future__ import annotations + +from workers.adaptive_router import AdaptiveRouter, MARKET_SOURCES, plan_for_market + + +class TestPlanForMarket: + def test_returns_the_full_static_plan_when_nothing_is_blocked(self): + router = AdaptiveRouter() + assert plan_for_market("GB", router=router) == MARKET_SOURCES["GB"] + + def test_unknown_market_falls_back_to_gleif_only(self): + router = AdaptiveRouter() + assert plan_for_market("ZZ", router=router) == ["gleif"] + + def test_defaults_to_the_process_wide_router_when_none_is_passed(self): + # Just proves the call doesn't blow up and returns a real plan -- + # not asserting anything about global state here (other tests use + # their own injected router precisely to avoid cross-test pollution + # of the shared default_router). + assert plan_for_market("FI") == MARKET_SOURCES["FI"] + + +class TestBlockTracking: + def test_a_freshly_blocked_source_is_excluded_from_the_plan(self): + router = AdaptiveRouter() + router.record_blocked("GB", "companies_house") + plan = plan_for_market("GB", router=router) + assert "companies_house" not in plan + assert "gleif" in plan # only the blocked source is removed, not the whole market + + def test_is_blocked_is_false_before_any_block_is_recorded(self): + router = AdaptiveRouter() + assert router.is_blocked("GB", "companies_house") is False + + def test_blocking_is_scoped_per_market_not_global_to_the_source_id(self): + # gleif appears in every market's plan -- blocking it for GB must not + # silently remove it from FI/SE/etc too. + router = AdaptiveRouter() + router.record_blocked("GB", "gleif") + assert "gleif" not in plan_for_market("GB", router=router) + assert "gleif" in plan_for_market("FI", router=router) + assert "gleif" in plan_for_market("SE", router=router) + + def test_cooldown_expires_and_the_source_becomes_eligible_again(self): + router = AdaptiveRouter(cooldown_seconds=60.0) + router.record_blocked("GB", "companies_house", now=1000.0) + assert router.is_blocked("GB", "companies_house", now=1030.0) is True # still within cooldown + assert router.is_blocked("GB", "companies_house", now=1061.0) is False # cooldown elapsed + # And the plan reflects that: + assert "companies_house" in plan_for_market("GB", router=router) + + def test_clear_forgets_every_recorded_block(self): + router = AdaptiveRouter() + router.record_blocked("GB", "companies_house") + router.record_blocked("NO", "brreg") + router.clear() + assert plan_for_market("GB", router=router) == MARKET_SOURCES["GB"] + assert plan_for_market("NO", router=router) == MARKET_SOURCES["NO"] + + def test_two_router_instances_never_share_state(self): + # Regression guard for a shared-mutable-default bug class: each + # AdaptiveRouter() must own its own dict, not a class-level one. + a = AdaptiveRouter() + b = AdaptiveRouter() + a.record_blocked("GB", "companies_house") + assert a.is_blocked("GB", "companies_house") is True + assert b.is_blocked("GB", "companies_house") is False diff --git a/services/api/tests/test_connectors_blocked.py b/services/api/tests/test_connectors_blocked.py new file mode 100644 index 0000000..42e0d13 --- /dev/null +++ b/services/api/tests/test_connectors_blocked.py @@ -0,0 +1,104 @@ +""" +P014: every connector used by pipeline.py's adaptive plan (gleif, brreg, +companies_house, prh, cvr) used to collapse EVERY non-200 response into a +bare `None`, indistinguishable from "record genuinely doesn't exist". This +tests that the block-status subset (401/403/429/451) now raises +SourceBlockedError instead, while every other non-200 status (404, 500, ...) +still returns None exactly as before -- this is an ADDITIVE signal, not a +behavior change for the non-block case. +""" +from __future__ import annotations + +import httpx +import pytest +import respx + +from sources.connectors import ( + SourceBlockedError, + brreg_by_orgnr, + companies_house_by_number, + cvr_by_number, + gleif_by_lei, + prh_by_businessid, +) + +GLEIF_URL = "https://api.gleif.org/api/v1/lei-records/5493001KJTIIGC8Y1R12" +BRREG_URL = "https://data.brreg.no/enhetsregisteret/api/enheter/923609016" +COMPANIES_HOUSE_URL = "https://api.company-information.service.gov.uk/company/00000006" +PRH_URL = "https://avoindata.prh.fi/opendata-ytj-api/v3/companies?businessId=2077474-0" +CVR_URL = "https://cvrapi.dk/api?search=10150817&country=dk" + + +@pytest.mark.asyncio +@respx.mock +@pytest.mark.parametrize("status", [401, 403, 429, 451]) +async def test_gleif_raises_source_blocked_on_block_statuses(status): + respx.get(GLEIF_URL).mock(return_value=httpx.Response(status, json={})) + with pytest.raises(SourceBlockedError) as exc_info: + await gleif_by_lei("5493001KJTIIGC8Y1R12") + assert exc_info.value.source_id == "gleif" + assert exc_info.value.status_code == status + + +@pytest.mark.asyncio +@respx.mock +async def test_gleif_a_404_still_returns_none_not_blocked(monkeypatch): + respx.get(GLEIF_URL).mock(return_value=httpx.Response(404, json={})) + result = await gleif_by_lei("5493001KJTIIGC8Y1R12") + assert result is None # unchanged pre-P014 behavior for a genuine "not found" + + +@pytest.mark.asyncio +@respx.mock +async def test_brreg_raises_source_blocked_on_429(): + respx.get(BRREG_URL).mock(return_value=httpx.Response(429, json={})) + with pytest.raises(SourceBlockedError) as exc_info: + await brreg_by_orgnr("923609016") + assert exc_info.value.source_id == "brreg" + + +@pytest.mark.asyncio +@respx.mock +async def test_companies_house_raises_source_blocked_on_403(monkeypatch): + monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key") + respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(403, json={})) + with pytest.raises(SourceBlockedError) as exc_info: + await companies_house_by_number("00000006") + assert exc_info.value.source_id == "companies_house" + + +@pytest.mark.asyncio +async def test_companies_house_missing_key_still_delivers_honest_level_0_not_a_block(): + # No COMPANIES_HOUSE_API_KEY set -- this is a config gap, not a source + # block, and must keep returning the honest "unavailable" result it + # already did (never raise SourceBlockedError for a problem on OUR side). + result = await companies_house_by_number("00000006") + assert result is not None + assert result.provenance[0].level == 0 + assert result.provenance[0].transform == "unavailable_no_key" + + +@pytest.mark.asyncio +@respx.mock +async def test_prh_raises_source_blocked_on_401(): + respx.get(PRH_URL).mock(return_value=httpx.Response(401, json={})) + with pytest.raises(SourceBlockedError) as exc_info: + await prh_by_businessid("2077474-0") + assert exc_info.value.source_id == "prh" + + +@pytest.mark.asyncio +@respx.mock +async def test_cvr_raises_source_blocked_on_451(): + respx.get(CVR_URL).mock(return_value=httpx.Response(451, json={})) + with pytest.raises(SourceBlockedError) as exc_info: + await cvr_by_number("10150817") + assert exc_info.value.source_id == "cvr" + + +@pytest.mark.asyncio +@respx.mock +async def test_cvr_a_500_still_returns_none_not_blocked(): + respx.get(CVR_URL).mock(return_value=httpx.Response(500, json={})) + result = await cvr_by_number("10150817") + assert result is None diff --git a/services/api/tests/test_orchestrator_run_idempotent.py b/services/api/tests/test_orchestrator_run_idempotent.py new file mode 100644 index 0000000..6058821 --- /dev/null +++ b/services/api/tests/test_orchestrator_run_idempotent.py @@ -0,0 +1,112 @@ +""" +P014: services/api/orchestrator/main.py had ZERO test coverage before this +pass, and `POST /orders/{oid}/run` had no idempotency protection at all -- +calling it twice for the same order re-ran the full compliant sourcing +pipeline a second time, silently overwriting `_JOBS[oid]` with a second, +independently-produced delivery. This is the actual "duplicate delivery" +gap P014's recon flagged NOT MET (nothing downstream of order creation was +idempotent). These tests exercise the real ASGI app end to end (httpx +ASGITransport, no mocked HTTP layer) and prove: a sequential retry returns +the cached job without re-running the pipeline, a TRUE CONCURRENT race +(two requests in flight at once) is also closed by the per-order asyncio +lock, and an unknown order still 404s exactly as before. +""" +from __future__ import annotations + +import asyncio + +import httpx +import pytest + +from orchestrator.main import app + + +def _fake_job_result(calls: list[str], order_id: str) -> dict: + calls.append(order_id) + return { + "stage": "qa", + "pct": 100, + "state": "done", + "metrics": {"rows": 1, "dropped_below_threshold": 0, "per_market": {"NO": 1}, "avg_confidence": 0.95}, + "records": [{"market": "NO", "name": "Test AS", "confidence": 0.95}], + } + + +@pytest.mark.asyncio +async def test_sequential_retry_returns_cached_job_without_rerunning_pipeline(monkeypatch): + calls: list[str] = [] + + async def fake_run_job(order_id: str, config: dict) -> dict: + return _fake_job_result(calls, order_id) + + monkeypatch.setattr("orchestrator.main.run_job", fake_run_job) + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + created = await client.post("/orders", json={"flow": "custom", "config": {"markets": ["NO"]}}) + oid = created.json()["order_id"] + + r1 = await client.post(f"/orders/{oid}/run") + r2 = await client.post(f"/orders/{oid}/run") + r3 = await client.post(f"/orders/{oid}/run") + + assert r1.status_code == 200 + assert r1.json() == r2.json() == r3.json() + assert calls == [oid] # run_job invoked exactly once across three /run calls + + +@pytest.mark.asyncio +async def test_true_concurrent_race_still_invokes_run_job_exactly_once(monkeypatch): + calls: list[str] = [] + + async def fake_run_job(order_id: str, config: dict) -> dict: + # Widen the race window: both requests reach this await before + # either has written to _JOBS, which is exactly the race a bare + # `if oid in _JOBS` check (no lock) would lose. + await asyncio.sleep(0.05) + return _fake_job_result(calls, order_id) + + monkeypatch.setattr("orchestrator.main.run_job", fake_run_job) + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + created = await client.post("/orders", json={"flow": "custom", "config": {"markets": ["NO"]}}) + oid = created.json()["order_id"] + + r1, r2 = await asyncio.gather( + client.post(f"/orders/{oid}/run"), + client.post(f"/orders/{oid}/run"), + ) + + assert r1.status_code == 200 and r2.status_code == 200 + assert r1.json() == r2.json() + assert calls == [oid] # THE regression this test guards: exactly one real pipeline run + + +@pytest.mark.asyncio +async def test_run_on_unknown_order_still_404s(): + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.post("/orders/does-not-exist/run") + assert res.status_code == 404 + + +@pytest.mark.asyncio +async def test_different_orders_each_get_their_own_real_run(monkeypatch): + calls: list[str] = [] + + async def fake_run_job(order_id: str, config: dict) -> dict: + return _fake_job_result(calls, order_id) + + monkeypatch.setattr("orchestrator.main.run_job", fake_run_job) + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + o1 = (await client.post("/orders", json={"flow": "custom", "config": {"markets": ["NO"]}})).json()["order_id"] + o2 = (await client.post("/orders", json={"flow": "custom", "config": {"markets": ["GB"]}})).json()["order_id"] + + await client.post(f"/orders/{o1}/run") + await client.post(f"/orders/{o2}/run") + + # Idempotency is per-order, not global -- two distinct orders both really run. + assert sorted(calls) == sorted([o1, o2]) diff --git a/services/api/tests/test_pipeline_source_blocked.py b/services/api/tests/test_pipeline_source_blocked.py new file mode 100644 index 0000000..0732b33 --- /dev/null +++ b/services/api/tests/test_pipeline_source_blocked.py @@ -0,0 +1,114 @@ +""" +P014: integration proof that a source signaling a block (HTTP 401/403/429/451) +during a real run_job() call (a) never crashes or fabricates a result, (b) is +recorded on the process-wide adaptive_router so the engine backs off that +source, and (c) the pipeline still delivers from the next compliant source in +the plan -- exactly the "source-racing several authorized sources" behavior +the siax-0pt skill documents, now with an honest memory of blocks instead of +silently re-trying a rate-limited source with none. + +Uses the real default_router (module-level singleton pipeline.py actually +uses), cleared before and after every test in this file so this suite never +leaks state into -- or picks up state from -- any other test module that +imports workers.pipeline. +""" +from __future__ import annotations + +import httpx +import pytest +import respx + +from workers.adaptive_router import default_router +from workers.pipeline import run_job + +COMPANIES_HOUSE_URL = "https://api.company-information.service.gov.uk/company/00000006" +GLEIF_URL = "https://api.gleif.org/api/v1/lei-records/00000006" + + +@pytest.fixture(autouse=True) +def _clean_router(): + default_router.clear() + yield + default_router.clear() + + +class TestSourceBlockedDuringARealRun: + @pytest.mark.asyncio + @respx.mock + async def test_a_429_from_one_source_is_recorded_as_blocked_and_the_run_still_delivers_from_the_next_source( + self, monkeypatch + ): + monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key") + respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(429, json={"error": "rate limited"})) + respx.get(GLEIF_URL).mock( + return_value=httpx.Response( + 200, + json={"data": {"attributes": {"entity": {"legalName": {"name": "Fallback Ltd"}, "status": "ACTIVE", "legalAddress": {}}}}}, + ) + ) + + assert default_router.is_blocked("GB", "companies_house") is False # sanity: clean slate + + result = await run_job("order_block_1", {"markets": ["GB"], "confidence": 50, "seeds": {"GB": ["00000006"]}}) + + assert result["state"] == "done" + assert result["metrics"]["rows"] == 1 + assert result["records"][0]["name"] == "Fallback Ltd" + assert result["records"][0]["source"] == "gleif" # companies_house was blocked, gleif answered + + # THE regression this test guards: the block was actually recorded, + # not silently discarded the way a bare `except Exception: res = None` + # would (that's still the fallback for genuine errors, but a block is + # now distinguishable from one). + assert default_router.is_blocked("GB", "companies_house") is True + + @pytest.mark.asyncio + @respx.mock + async def test_a_blocked_source_is_excluded_from_the_plan_on_the_very_next_key_in_the_same_run(self, monkeypatch): + monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key") + gb_second_url = "https://api.company-information.service.gov.uk/company/11111111" + gleif_second_url = "https://api.gleif.org/api/v1/lei-records/11111111" + + companies_house_calls = respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(429, json={})) + # No mock registered for gb_second_url at all -- if the pipeline tried + # companies_house AGAIN for the second key after it was just blocked, + # respx would raise on the unmocked request and this test would fail + # with a connection error, not a clean assertion. That absence IS the + # assertion: the plan excluded the just-blocked source. + respx.get(GLEIF_URL).mock( + return_value=httpx.Response(200, json={"data": {"attributes": {"entity": {"legalName": {"name": "First"}, "legalAddress": {}}}}}) + ) + respx.get(gleif_second_url).mock( + return_value=httpx.Response(200, json={"data": {"attributes": {"entity": {"legalName": {"name": "Second"}, "legalAddress": {}}}}}) + ) + + result = await run_job( + "order_block_2", + {"markets": ["GB"], "confidence": 50, "seeds": {"GB": ["00000006", "11111111"]}}, + ) + + assert result["state"] == "done" + assert result["metrics"]["rows"] == 2 + names = {r["name"] for r in result["records"]} + assert names == {"First", "Second"} + # companies_house was hit exactly once (for the first key, where it + # got blocked) -- never retried for the second key in this run. + assert companies_house_calls.call_count == 1 + + @pytest.mark.asyncio + @respx.mock + async def test_a_plain_5xx_is_not_treated_as_a_block(self, monkeypatch): + # A genuine server error is a different signal from a compliance + # block -- it should NOT be recorded via record_blocked (that would + # incorrectly suppress a source that's just having a bad moment, not + # one that told the engine to stop). + monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key") + respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(500, json={})) + respx.get(GLEIF_URL).mock( + return_value=httpx.Response(200, json={"data": {"attributes": {"entity": {"legalName": {"name": "X"}, "legalAddress": {}}}}}) + ) + + result = await run_job("order_block_3", {"markets": ["GB"], "confidence": 50, "seeds": {"GB": ["00000006"]}}) + + assert result["state"] == "done" + assert default_router.is_blocked("GB", "companies_house") is False diff --git a/services/api/workers/adaptive_router.py b/services/api/workers/adaptive_router.py index 5b66c70..902ab20 100644 --- a/services/api/workers/adaptive_router.py +++ b/services/api/workers/adaptive_router.py @@ -1,6 +1,40 @@ -"""Adaptive router — all methods, no compliance restrictions. See b0ts for scraping.""" +""" +Adaptive router — per-market compliant-source planner. + +P014 CORRECTION: this file's previous docstring read "all methods, no +compliance restrictions. See b0ts for scraping." That phrasing directly +contradicts this product's own compliance stance (CLAUDE.md §1: official +API/bulk -> authorized public web -> licensed partner data -> otherwise +level 0, never a bypass) and, worse, pointed a future reader at `b0ts` -- a +SEPARATE repo that does non-compliant scraping and is out of scope/forbidden +to touch from here. Nothing in this file ever called b0ts or any bypass +path; the line was simply wrong and is corrected, not extended. + +What this file actually does: `plan_for_market(market)` returns the ordered +list of registered source_ids `pipeline.py._fetch` tries, in priority order, +for a given market. Every entry is a name pipeline.py may or may not have an +implemented connector for (see sources/connectors.py's CONNECTORS dict) -- +an unimplemented source_id is simply skipped by `_fetch`, it is never a +license to build a bypass in its place. + +P014 ADDITION: `record_blocked` / `is_blocked` give this planner a memory of +which (market, source_id) pairs recently told the engine to stop (HTTP +401/403/429/451 -- see sources/connectors.py's SourceBlockedError). Before +this pass, a source that started rate-limiting or blocking the engine mid-run +was retried on every subsequent record with no memory at all -- not a +compliance bypass, but not "respecting the block as a stop signal" either +(the same principle 0pt-motor's AdaptiveRouter documents for its own, +separate method-tier selector). `plan_for_market` now filters a currently- +blocked source out of the plan for the rest of the block's cooldown window, +so the engine backs off instead of hammering it. This is process-local, +in-memory state (matches every other piece of in-memory state this engine +documents honestly elsewhere, e.g. orchestrator/main.py's _ORDERS/_JOBS) -- +not a persistent, cross-process reputation system. +""" from __future__ import annotations +import time + MARKET_SOURCES: dict[str, list[str]] = { "NO": ["brreg", "brreg_under", "proff", "1881", "gulesider", "flowfirma", "selskapssok", "regnskapsbasen", "okonomi", "visverdi", "gleif"], "GB": ["companies_house", "gleif"], @@ -10,6 +44,54 @@ MARKET_SOURCES: dict[str, list[str]] = { "SE": ["gleif"], } -def plan_for_market(market: str) -> list[str]: - """Return the source plan for a market. Deep scraping via b0ts service.""" - return MARKET_SOURCES.get(market, ["gleif"]) +# How long a (market, source_id) pair is excluded from plan_for_market after +# being reported blocked. A fixed cooldown, not exponential backoff -- this +# is a planner, not a retry/backoff controller; a source is free to be tried +# again on the next run once the window elapses. +DEFAULT_BLOCK_COOLDOWN_SECONDS = 300.0 + + +class AdaptiveRouter: + """Tracks which (market, source_id) pairs are currently in a block cooldown.""" + + def __init__(self, cooldown_seconds: float = DEFAULT_BLOCK_COOLDOWN_SECONDS): + self._cooldown = cooldown_seconds + self._blocked_until: dict[tuple[str, str], float] = {} + + def record_blocked(self, market: str, source_id: str, *, now: float | None = None) -> None: + """Mark (market, source_id) as blocked starting now.""" + self._blocked_until[(market, source_id)] = (now if now is not None else time.monotonic()) + self._cooldown + + def is_blocked(self, market: str, source_id: str, *, now: float | None = None) -> bool: + deadline = self._blocked_until.get((market, source_id)) + if deadline is None: + return False + current = now if now is not None else time.monotonic() + if current >= deadline: + # Cooldown elapsed -- clear it so the map doesn't grow forever. + del self._blocked_until[(market, source_id)] + return False + return True + + def clear(self) -> None: + """Test/process-reset hook -- forgets every recorded block.""" + self._blocked_until.clear() + + +# Process-wide singleton, mirroring the in-memory-state convention already +# used elsewhere in this engine (e.g. orchestrator/main.py's _ORDERS/_JOBS). +# Injectable in `plan_for_market` for tests rather than only reachable via +# global mutable state. +default_router = AdaptiveRouter() + + +def plan_for_market(market: str, *, router: AdaptiveRouter | None = None) -> list[str]: + """ + Return the ordered source plan for a market, minus any source_id + currently in a block cooldown for that market. `router` defaults to the + process-wide `default_router`; tests inject their own instance so they + never share state with each other or with production. + """ + r = router if router is not None else default_router + plan = MARKET_SOURCES.get(market, ["gleif"]) + return [source_id for source_id in plan if not r.is_blocked(market, source_id)] diff --git a/services/api/workers/pipeline.py b/services/api/workers/pipeline.py index c315893..b6da183 100644 --- a/services/api/workers/pipeline.py +++ b/services/api/workers/pipeline.py @@ -29,13 +29,14 @@ from workers.quality_bridge import run_quality_bridge from sources.connectors import ( CompanyResult, + SourceBlockedError, brreg_by_orgnr, companies_house_by_number, cvr_by_number, gleif_by_lei, prh_by_businessid, ) -from workers.adaptive_router import plan_for_market +from workers.adaptive_router import default_router, plan_for_market logger = logging.getLogger(__name__) @@ -78,12 +79,20 @@ async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]: # 1) sourcing + 2) extraction for market in markets: keys = seeds.get(market, []) - plan = plan_for_market(market) got = 0 for key in keys: + # Recomputed per key (not hoisted above this loop): a source + # blocked while fetching an earlier key in this same market must + # be excluded from the plan for every key still to come in this + # run, not just retried forever with no memory (P014 -- see + # adaptive_router.py's module docstring). + plan = plan_for_market(market) for source_id in plan: try: res = await _fetch(source_id, key) + except SourceBlockedError: + default_router.record_blocked(market, source_id) + res = None except Exception: res = None if res and res.name: diff --git a/sql/003_datasets.sql b/sql/003_datasets.sql new file mode 100644 index 0000000..6552d89 --- /dev/null +++ b/sql/003_datasets.sql @@ -0,0 +1,41 @@ +-- P014: fills a real, previously-undiscovered gap in the migration chain. +-- sql/004_iceberg_write_boundary.sql's own comment already documented (but +-- did not fix) that `opt_dataset_iceberg_mappings.dataset_id REFERENCES +-- opt_datasets (dataset_id)` points at a table NO sql/ file ever defined -- +-- meaning `psql -d opt < sql/001... < sql/002... < sql/004...` (exactly the +-- flow docs/runbook/backup-restore.md's own "Målbild" describes, and exactly +-- what a real restore drill executes) has never actually been runnable +-- end to end. This was found by running that exact sequence for real against +-- a throwaway Postgres container during P014's restore drill, not by +-- inspecting the code. +-- +-- Numbered 003 (the previously-unused gap between 002_source_registry.sql +-- and 004_iceberg_write_boundary.sql) so the existing ordering -- 001 -> 002 +-- -> 004 -> 005 -- becomes 001 -> 002 -> 003 -> 004 -> 005 without renumbering +-- anything already shipped. +-- +-- Columns mirror src/services/dataset-service.ts's `DatasetRecord` exactly +-- (that file's own doc comment: "Storage shape: nullable columns, exactly as +-- the eventual `opt_datasets` table would hold them") -- this is the real +-- application contract, not a guess. `dataset_type`/`status` CHECK +-- constraints mirror src/types/dataset.ts's `DatasetType`/`DatasetStatus` +-- unions verbatim. DatasetService itself remains in-memory-only in this +-- pass (wiring it to this table is a separate, larger change than closing +-- the migration-chain gap); this table exists so the SCHEMA a fresh +-- database gets by replaying sql/*.sql matches what 004/005 already assume. + +CREATE TABLE IF NOT EXISTS opt_datasets ( + dataset_id TEXT PRIMARY KEY, + tenant_id TEXT NOT NULL, + project_id TEXT NOT NULL, + name TEXT NOT NULL, + dataset_type TEXT NOT NULL CHECK (dataset_type IN ('raw', 'structured', 'enriched', 'curated', 'feature', 'derived')), + status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'archived', 'deprecated')), + description TEXT, + correlation_id TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ, + created_by TEXT +); + +CREATE INDEX IF NOT EXISTS opt_datasets_tenant_project_idx ON opt_datasets (tenant_id, project_id); diff --git a/sql/006_orders_idempotency.sql b/sql/006_orders_idempotency.sql new file mode 100644 index 0000000..075e54f --- /dev/null +++ b/sql/006_orders_idempotency.sql @@ -0,0 +1,24 @@ +-- P014: duplicate-delivery idempotency for POST /api/orders (docs/04, §01). +-- lib/orders-store.ts previously had NO idempotency mechanism at all -- a +-- retried POST (a prospect double-clicking "submit", a client retrying on a +-- dropped response, a proxy replaying a request) created a second, distinct +-- `orders` row for what was semantically the same submission every time, +-- with no way to detect or collapse the duplicate afterward (P014 recon: +-- "Duplicate delivery idempotent: NOT MET"). +-- +-- `idempotency_key` is caller-supplied (the standard `Idempotency-Key` +-- header convention this skill's own docs already name as the expected +-- 0pt/orders convention) and OPTIONAL: an order created with no key behaves +-- exactly as before (multiple distinct rows on retry) -- this migration adds +-- a mechanism, it does not retroactively require every caller to use it. +-- When present, it must be unique across ALL orders (not scoped per-org): +-- `/api/orders` POST is intentionally anonymous (a prospect can draft/quote +-- before signing in -- see app/api/orders/route.ts), so there is no org_id +-- to scope by at creation time; a caller-generated key (e.g. a UUID minted +-- once per logical submission) is expected to be globally unique on its own, +-- matching the standard Stripe-style idempotency-key contract. +ALTER TABLE orders ADD COLUMN IF NOT EXISTS idempotency_key text; + +CREATE UNIQUE INDEX IF NOT EXISTS orders_idempotency_key_uq + ON orders (idempotency_key) + WHERE idempotency_key IS NOT NULL; diff --git a/src/auth/internal-token.ts b/src/auth/internal-token.ts new file mode 100644 index 0000000..852d5b9 --- /dev/null +++ b/src/auth/internal-token.ts @@ -0,0 +1,73 @@ +/** + * `src/auth/internal-token.ts` -- verification for the ONE machine-to-machine + * credential 0pt already uses for engine (Python) -> edge (Next.js) calls: + * the `X-Internal-Token` header, checked against the `INTERNAL_TOKEN` env var + * (the same shared secret `lib/engine.ts` and `lib/health-checks.ts` already + * send on the opposite-direction edge -> engine calls). + * + * P014 FIX: `services/api/workers/quality_bridge.py` has always called + * `POST /api/0pt/quality/bridge-run` with `X-Internal-Token`, never an + * `Authorization: Bearer` header. But this route sits under + * `/api/0pt/:path*`, which `middleware.ts` guards unconditionally with real + * Zitadel Bearer verification -- and nothing anywhere checked + * `X-Internal-Token` as an alternative credential. The result: in any + * environment with the default `AUTH_ENFORCE` posture (enforced), the + * Python engine's ONLY call into this bridge was rejected with 401 by + * middleware before the route handler ever ran. `OPT_QUALITY_BRIDGE_ENABLED=true` + * therefore could never actually succeed against a real deployment -- + * quality_bridge.py's fail-open design (see its own docstring) silently + * absorbed the 401 as `status: "unavailable"`, so nothing outwardly broke, + * but the whole point of building the bridge (an advisory second opinion + * from the TS QualityService) was unreachable. Neither test suite caught + * this: the TS test (tests/p0/quality-bridge.test.ts) calls the route's + * POST function directly, bypassing middleware.ts entirely; the Python test + * (services/api/tests/test_quality_bridge.py) mocks the HTTP layer and never + * exercises the edge's real auth gate. + * + * This module verifies `X-Internal-Token` in constant time. middleware.ts + * accepts it ONLY for the one route the engine actually calls + * (`/api/0pt/quality/bridge-run`) -- not a blanket bypass for the rest of + * `/api/0pt/*`. Fails closed: an unset/empty `INTERNAL_TOKEN` env var never + * authenticates, even against an empty or missing header (no accidental + * "unset equals unset, so let it through"). + * + * Uses only Web APIs (`TextEncoder`), not `node:crypto` -- Next.js + * middleware runs on the Edge runtime, which does not expose Node's + * `crypto.timingSafeEqual`. + */ + +/** Minimal header accessor -- Web `Headers` and `NextRequest.headers` satisfy this. */ +export interface HeaderLookup { + get(name: string): string | null | undefined; +} + +export const INTERNAL_TOKEN_HEADER = "x-internal-token"; + +/** Constant-time-ish byte comparison (Edge-runtime safe: no node:crypto). */ +function timingSafeEqualStr(a: string, b: string): boolean { + const enc = new TextEncoder(); + const aBytes = enc.encode(a); + const bBytes = enc.encode(b); + const len = Math.max(aBytes.length, bBytes.length); + let diff = aBytes.length ^ bBytes.length; + for (let i = 0; i < len; i++) { + diff |= (aBytes[i] ?? 0) ^ (bBytes[i] ?? 0); + } + return diff === 0; +} + +/** + * True only when `X-Internal-Token` is present AND matches a non-empty + * configured `INTERNAL_TOKEN`. Fails closed on every other combination + * (missing header, missing/empty env var, mismatch). + */ +export function verifyInternalServiceToken( + headers: HeaderLookup, + env: { INTERNAL_TOKEN?: string } = process.env, +): boolean { + const expected = env.INTERNAL_TOKEN?.trim(); + if (!expected) return false; + const provided = headers.get(INTERNAL_TOKEN_HEADER); + if (!provided) return false; + return timingSafeEqualStr(provided, expected); +} diff --git a/src/http/tenant-guard.ts b/src/http/tenant-guard.ts new file mode 100644 index 0000000..9892f72 --- /dev/null +++ b/src/http/tenant-guard.ts @@ -0,0 +1,68 @@ +/** + * src/http/tenant-guard.ts + * ========================== + * P014 fix: every route under `app/api/0pt/**` sits behind `middleware.ts`, + * which performs real Zitadel Bearer verification and — only when a + * principal was actually verified — forwards the result as the + * `x-0pt-verified-tenant` request header (see middleware.ts, src/auth/context.ts). + * + * Independently of that, every one of these route handlers reads `tenant_id` + * from CLIENT-CONTROLLED input (a `?tenant_id=` query param or a JSON body + * field) to decide which tenant's rows to read or write. Until this pass, + * NONE of them cross-checked that client-supplied value against the verified + * header — so a caller holding a valid token for tenant A could read or + * write tenant B's sources, rights, provenance, health, quality rules/runs, + * documents, or lakehouse mappings simply by naming `tenant_id=t_b` in the + * request. `SourceRegistryService.assertSourceTenant` (and friends) only + * checks that the object being read *belongs* to the tenant_id it was asked + * to check against — it has no way to know whether the CALLER is actually + * entitled to assert that tenant_id in the first place. That entitlement + * check belongs at the HTTP boundary, once, here. + * + * This mirrors the exact decision `src/auth/context.ts`'s `resolveAuthContext` + * already makes for the enforced-Bearer path ("a disagreeing X-Tenant-Id + * header is a 403"), applied uniformly to every 0pt capability route that + * takes tenant_id as a body/query parameter rather than relying on + * middleware to inject ctx.tenantId directly (most of these routes predate + * that plumbing and read tenant_id straight from the request; fixing that + * more deeply is a larger refactor than this pass — this closes the actual + * authz hole without changing every route's request contract). + */ +import { NextResponse, type NextRequest } from "next/server"; + +export const VERIFIED_TENANT_HEADER = "x-0pt-verified-tenant"; + +/** + * Returns a 403 NextResponse when `tenantId` (whatever the route read it + * from — query string or body) disagrees with the tenant middleware.ts + * already verified for this request. Returns `null` when the caller may + * proceed — either because the tenant matches, or because no verified + * tenant is present on the request at all. + * + * The header is absent only in the explicit local-development escape hatch + * (`AUTH_ENFORCE=false` with no bearer token presented) — `assertAuthConfigSafe()` + * refuses to boot that way in production, so an absent header here can only + * happen in a mode that already accepts fully unauthenticated requests. + * Rejecting in that mode would be strictly more restrictive than the + * existing dev escape hatch intends, so this deliberately no-ops rather than + * invents a tenant to compare against. + */ +export function assertTenantMatchesVerified( + request: NextRequest, + tenantId: string | null | undefined, +): NextResponse | null { + const verified = request.headers.get(VERIFIED_TENANT_HEADER); + if (!verified) return null; + if (!tenantId) return null; // handlers already 400 on a missing tenant_id separately + if (tenantId !== verified) { + return NextResponse.json( + { + error: "OPT_ACCESS_DENIED", + message: "tenant_id disagrees with the authenticated token's tenant", + details: { requested_tenant: tenantId, verified_tenant: verified }, + }, + { status: 403 }, + ); + } + return null; +} diff --git a/src/services/dataset-iceberg-mapping-service.ts b/src/services/dataset-iceberg-mapping-service.ts index 4324bcd..65702ed 100644 --- a/src/services/dataset-iceberg-mapping-service.ts +++ b/src/services/dataset-iceberg-mapping-service.ts @@ -125,9 +125,21 @@ export class DatasetIcebergMappingService { return this.toMapping(record); } - async getMapping(mappingId: string): Promise { + // P014: `tenantId` is optional so trusted in-process callers (existing + // unit tests, and internal callers that already resolved the tenant some + // other way) are unaffected -- but both HTTP routes that reach this method + // now always pass it, closing a real gap: `getMapping` used to be + // reachable by mapping ID alone with NO tenant check anywhere, so any + // authenticated caller could read -- or, via requestIcebergWrite below, + // WRITE a new snapshot onto -- another tenant's dataset<->Iceberg-table + // mapping just by guessing/enumerating dataset_iceberg_mapping_id. A + // mismatched tenantId reports identically to "not found" (no existence + // oracle), matching SourceRegistryService.assertSourceTenant's convention. + async getMapping(mappingId: string, tenantId?: string): Promise { const record = this.mappings.get(mappingId); - if (!record) throw new Error(`Dataset Iceberg mapping ${mappingId} not found`); + if (!record || (tenantId && record.tenant_id !== tenantId)) { + throw new Error(`Dataset Iceberg mapping ${mappingId} not found`); + } return this.toMapping(record); } diff --git a/src/services/document-extraction-service.ts b/src/services/document-extraction-service.ts index 003bcd0..c70c4a9 100644 --- a/src/services/document-extraction-service.ts +++ b/src/services/document-extraction-service.ts @@ -249,7 +249,22 @@ export class DocumentExtractionService { return record; } - async listExtractionsForDocument(documentId: string): Promise { + // P014: `tenantId` is optional so existing trusted in-process callers are + // unaffected, but the HTTP route now always supplies it -- closing a real + // gap: this used to be reachable by documentId alone with NO tenant + // check anywhere, so any authenticated caller could list another + // tenant's OCR/structuring extraction results (potentially containing + // extracted document content, e.g. invoice fields) by guessing or + // enumerating documentId. A mismatched tenantId reports identically to + // "not found" (no existence oracle), same convention as the rest of the + // surface. + async listExtractionsForDocument(documentId: string, tenantId?: string): Promise { + if (tenantId) { + const document = await this.captureService.getDocument(documentId).catch(() => undefined); + if (!document || document.tenant_id !== tenantId) { + throw new Error(`Document ${documentId} not found`); + } + } return Array.from(this.extractions.values()).filter((r) => r.document_id === documentId); } } diff --git a/src/services/iceberg-write-service.ts b/src/services/iceberg-write-service.ts index 353aa4a..e938b0e 100644 --- a/src/services/iceberg-write-service.ts +++ b/src/services/iceberg-write-service.ts @@ -65,6 +65,15 @@ export interface MapDatasetToIcebergTableResult { export interface RequestIcebergWriteInput { dataset_iceberg_mapping_id: string; + /** + * P014: optional so existing trusted in-process callers are unaffected, + * but the HTTP route (app/api/0pt/lakehouse/mappings/[mappingId]/writes) + * always supplies it now. When present, it must match the mapping's own + * tenant_id or the write is refused as not-found -- see getMapping's + * doc comment for the gap this closes (a foreign mapping_id used to let + * any caller register a snapshot against another tenant's table). + */ + tenant_id?: string; actor_id: string; correlation_id: string; snapshot_type?: TableSnapshotType; @@ -176,7 +185,7 @@ export class IcebergWriteService { } async requestIcebergWrite(input: RequestIcebergWriteInput): Promise { - const mapping = await this.mappingService.getMapping(input.dataset_iceberg_mapping_id); + const mapping = await this.mappingService.getMapping(input.dataset_iceberg_mapping_id, input.tenant_id); if (mapping.mapping_status !== "active") { throw new Error(`Dataset Iceberg mapping ${input.dataset_iceberg_mapping_id} is not active (status=${mapping.mapping_status})`); } diff --git a/src/services/source-rate-policy-service.ts b/src/services/source-rate-policy-service.ts index 2e53bab..192abb6 100644 --- a/src/services/source-rate-policy-service.ts +++ b/src/services/source-rate-policy-service.ts @@ -103,9 +103,20 @@ export class SourceRatePolicyService { return this.toPolicy(record); } - async getSourceRatePolicy(sourceRatePolicyId: string): Promise { + // P014: `tenantId` is optional so existing internal callers that already + // resolved a policy from a tenant-scoped list (or are trusted, in-process + // callers) are unaffected — but the HTTP route now always passes it, which + // closes a real gap: this lookup used to be reachable by rate-policy ID + // alone with NO tenant check anywhere, so any authenticated caller could + // read another tenant's crawl-budget/rate-limit/compliance-notes record + // just by guessing or enumerating IDs. A mismatched tenantId is reported + // identically to "not found" (no existence oracle), matching the + // convention SourceRegistryService.assertSourceTenant already established. + async getSourceRatePolicy(sourceRatePolicyId: string, tenantId?: string): Promise { const record = this.policies.get(sourceRatePolicyId); - if (!record) throw new Error(`Source rate policy ${sourceRatePolicyId} not found`); + if (!record || (tenantId && record.tenant_id !== tenantId)) { + throw new Error(`Source rate policy ${sourceRatePolicyId} not found`); + } return this.toPolicy(record); } diff --git a/src/validation/0pt-schemas.ts b/src/validation/0pt-schemas.ts index d202c58..eb42f9e 100644 --- a/src/validation/0pt-schemas.ts +++ b/src/validation/0pt-schemas.ts @@ -170,6 +170,13 @@ export const mapDatasetToIcebergTableSchema = z.object({ /** POST /api/0pt/lakehouse/mappings/[mappingId]/writes -- RequestIcebergWriteInput. */ export const requestIcebergWriteSchema = z.object({ + // P014: required so the route can both (a) cross-check it against the + // caller's verified tenant (assertTenantMatchesVerified) and (b) thread it + // into IcebergWriteService.requestIcebergWrite -> getMapping, which + // refuses the write as not-found when it disagrees with the mapping's own + // tenant_id. See getMapping's doc comment for the cross-tenant write gap + // this closes. + tenant_id: z.string().min(1), actor_id: z.string().min(1), correlation_id: z.string().min(1), snapshot_type: z.string().min(1).optional(), diff --git a/tests/api/0pt-cross-tenant-routes.test.ts b/tests/api/0pt-cross-tenant-routes.test.ts new file mode 100644 index 0000000..b873303 --- /dev/null +++ b/tests/api/0pt-cross-tenant-routes.test.ts @@ -0,0 +1,192 @@ +/** + * P014 regression: HTTP-level proof that every app/api/0pt/** route which + * takes tenant_id from client-controlled input (query string or JSON body) + * now rejects a request whose tenant_id disagrees with the tenant + * middleware.ts already verified for the caller (forwarded as + * `x-0pt-verified-tenant`). Before this pass, NONE of these routes checked + * that header at all -- a valid token for tenant A could read or write + * tenant B's sources, rate policies, quality rules, documents, or lakehouse + * mappings simply by naming tenant_id=B in the request. + * + * These call the route handlers directly with a hand-built NextRequest that + * carries `x-0pt-verified-tenant` -- exactly the shape middleware.ts hands + * the route once a real Bearer token has been verified (see + * tests/api/anonymous-routes-auth.test.ts for the same "call the handler + * directly, simulate what upstream already established" pattern used for + * the /api/orders and /api/admin/metrics fail-closed regressions). + */ +import { describe, it, expect } from "vitest"; +import { NextRequest } from "next/server"; +import { POST as sourcesPOST, GET as sourcesGET } from "@/app/api/0pt/sources/route"; +import { GET as sourceDescribeGET } from "@/app/api/0pt/sources/[sourceId]/route"; +import { GET as ratePolicyGET } from "@/app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route"; +import { GET as mappingsListGET, POST as mappingsPOST } from "@/app/api/0pt/lakehouse/mappings/route"; +import { GET as mappingGetGET } from "@/app/api/0pt/lakehouse/mappings/[mappingId]/route"; +import { POST as qualityRulesPOST } from "@/app/api/0pt/quality/rules/route"; +import { GET as documentsGET } from "@/app/api/0pt/documents/capture/route"; + +const VERIFIED = "x-0pt-verified-tenant"; + +function getReq(url: string, verifiedTenant = "t_a"): NextRequest { + return new NextRequest(new URL(url, "http://localhost"), { headers: { [VERIFIED]: verifiedTenant } }); +} +function postReq(url: string, body: unknown, verifiedTenant = "t_a"): NextRequest { + return new NextRequest(new URL(url, "http://localhost"), { + method: "POST", + headers: { "content-type": "application/json", [VERIFIED]: verifiedTenant }, + body: JSON.stringify(body), + }); +} + +describe("cross-tenant HTTP regression: /api/0pt/sources", () => { + it("POST rejects registering a source under a foreign tenant_id (403, never reaches the service)", async () => { + const res = await sourcesPOST( + postReq("/api/0pt/sources", { + source_id: "src_attack_1", + tenant_id: "t_victim", + project_id: "p1", + source_type: "api", + name: "attacker-named source", + }), + ); + expect(res.status).toBe(403); + const body = await res.json(); + expect(body.error).toBe("OPT_ACCESS_DENIED"); + }); + + it("POST succeeds when tenant_id matches the verified tenant", async () => { + const res = await sourcesPOST( + postReq("/api/0pt/sources", { + source_id: "src_ok_1", + tenant_id: "t_a", + project_id: "p1", + source_type: "api", + name: "legit source", + }), + ); + expect(res.status).toBe(201); + }); + + it("GET (list) rejects a foreign tenant_id query param (403)", async () => { + const res = await sourcesGET(getReq("/api/0pt/sources?tenant_id=t_victim", "t_a")); + expect(res.status).toBe(403); + }); + + it("GET (list) succeeds when the query tenant_id matches", async () => { + const res = await sourcesGET(getReq("/api/0pt/sources?tenant_id=t_a", "t_a")); + expect(res.status).toBe(200); + }); +}); + +describe("cross-tenant HTTP regression: /api/0pt/sources/[sourceId]", () => { + it("GET describe rejects a foreign tenant_id (403, before any existence check)", async () => { + const res = await sourceDescribeGET(getReq("/api/0pt/sources/src_whatever?tenant_id=t_victim", "t_a"), { + params: Promise.resolve({ sourceId: "src_whatever" }), + }); + expect(res.status).toBe(403); + }); +}); + +describe("cross-tenant HTTP regression: /api/0pt/sources/rate-policies/[id]", () => { + it("GET rejects a foreign tenant_id (403) -- this route had NO tenant check of any kind before P014", async () => { + const res = await ratePolicyGET(getReq("/api/0pt/sources/rate-policies/srp_whatever?tenant_id=t_victim", "t_a"), { + params: Promise.resolve({ sourceRatePolicyId: "srp_whatever" }), + }); + expect(res.status).toBe(403); + }); + + it("GET without tenant_id at all is a 400, not an unscoped read", async () => { + const res = await ratePolicyGET(getReq("/api/0pt/sources/rate-policies/srp_whatever", "t_a"), { + params: Promise.resolve({ sourceRatePolicyId: "srp_whatever" }), + }); + expect(res.status).toBe(400); + }); +}); + +describe("cross-tenant HTTP regression: /api/0pt/lakehouse/mappings", () => { + it("POST rejects mapping a dataset under a foreign tenant_id (403)", async () => { + const res = await mappingsPOST( + postReq( + "/api/0pt/lakehouse/mappings", + { + tenant_id: "t_victim", + project_id: "p1", + dataset_id: "ds_1", + dataset_iceberg_mapping_id: "dim_attack_1", + namespace: "ns", + table_name: "tbl", + storage_uri: "s3://x/y", + catalog_ref: "cat", + actor_id: "actor1", + correlation_id: "corr1", + }, + "t_a", + ), + ); + expect(res.status).toBe(403); + }); + + it("GET (list) requires tenant_id -- omitting it no longer returns every tenant's mappings (was: full leak)", async () => { + const res = await mappingsListGET(getReq("/api/0pt/lakehouse/mappings", "t_a")); + expect(res.status).toBe(400); + }); + + it("GET (list) rejects a foreign tenant_id query param (403)", async () => { + const res = await mappingsListGET(getReq("/api/0pt/lakehouse/mappings?tenant_id=t_victim", "t_a")); + expect(res.status).toBe(403); + }); +}); + +describe("cross-tenant HTTP regression: /api/0pt/lakehouse/mappings/[mappingId]", () => { + it("GET rejects a foreign tenant_id (403) -- this route had NO tenant check of any kind before P014", async () => { + const res = await mappingGetGET( + getReq("/api/0pt/lakehouse/mappings/dim_whatever?tenant_id=t_victim", "t_a"), + { params: Promise.resolve({ mappingId: "dim_whatever" }) }, + ); + expect(res.status).toBe(403); + }); + + it("GET without tenant_id at all is a 400, not an unscoped read", async () => { + const res = await mappingGetGET(getReq("/api/0pt/lakehouse/mappings/dim_whatever", "t_a"), { + params: Promise.resolve({ mappingId: "dim_whatever" }), + }); + expect(res.status).toBe(400); + }); +}); + +describe("cross-tenant HTTP regression: /api/0pt/quality/rules", () => { + it("POST rejects creating a rule under a foreign tenant_id (403)", async () => { + const res = await qualityRulesPOST( + postReq( + "/api/0pt/quality/rules", + { + rule_id: "rule_attack_1", + tenant_id: "t_victim", + project_id: "p1", + dataset_id: "ds_1", + name: "attacker rule", + expectation: { expectation_type: "not_null", kwargs: { column: "name" } }, + }, + "t_a", + ), + ); + expect(res.status).toBe(403); + }); +}); + +describe("cross-tenant HTTP regression: /api/0pt/documents/capture (GET)", () => { + it("requires tenant_id -- omitting it no longer returns every tenant's captured documents (was: full leak)", async () => { + const res = await documentsGET(getReq("/api/0pt/documents/capture", "t_a")); + expect(res.status).toBe(400); + }); + + it("rejects a foreign tenant_id query param (403)", async () => { + const res = await documentsGET(getReq("/api/0pt/documents/capture?tenant_id=t_victim", "t_a")); + expect(res.status).toBe(403); + }); + + it("succeeds when the query tenant_id matches the verified tenant", async () => { + const res = await documentsGET(getReq("/api/0pt/documents/capture?tenant_id=t_a", "t_a")); + expect(res.status).toBe(200); + }); +}); diff --git a/tests/api/orders-idempotency.test.ts b/tests/api/orders-idempotency.test.ts new file mode 100644 index 0000000..754aefc --- /dev/null +++ b/tests/api/orders-idempotency.test.ts @@ -0,0 +1,62 @@ +/** + * P014: HTTP-level proof that POST /api/orders honors the `Idempotency-Key` + * header end to end (route -> lib/orders-store.ts), matching the pattern + * used elsewhere in this suite (tests/api/anonymous-routes-auth.test.ts) of + * calling the route handler directly with a hand-built NextRequest. + */ +import { describe, it, expect } from "vitest"; +import { NextRequest } from "next/server"; +import { POST as ordersPOST } from "@/app/api/orders/route"; + +function postReq(body: unknown, headers: Record = {}): NextRequest { + return new NextRequest(new URL("/api/orders", "http://localhost"), { + method: "POST", + headers: { "content-type": "application/json", ...headers }, + body: JSON.stringify(body), + }); +} + +const BODY = { flow: "custom", config: { markets: ["SE"], type: "Bolag", fieldIds: ["identity"], volume: 1000 } }; +const BODY_DIFFERENT = { flow: "custom", config: { markets: ["NO"], type: "Bolag", fieldIds: ["identity"], volume: 9000 } }; + +describe("POST /api/orders — Idempotency-Key", () => { + it("no header: two identical requests create two distinct orders (201, 201)", async () => { + const r1 = await ordersPOST(postReq(BODY)); + const r2 = await ordersPOST(postReq(BODY)); + expect(r1.status).toBe(201); + expect(r2.status).toBe(201); + const b1 = await r1.json(); + const b2 = await r2.json(); + expect(b1.order_id).not.toBe(b2.order_id); + }); + + it("REGRESSION: same Idempotency-Key replays the original order (201 then 200, same order_id)", async () => { + const key = `idem_${crypto.randomUUID()}`; + const r1 = await ordersPOST(postReq(BODY, { "Idempotency-Key": key })); + const r2 = await ordersPOST(postReq(BODY, { "Idempotency-Key": key })); + expect(r1.status).toBe(201); + expect(r2.status).toBe(200); // replayed, not created + const b1 = await r1.json(); + const b2 = await r2.json(); + expect(b2.order_id).toBe(b1.order_id); + expect(b2.total_ore).toBe(b1.total_ore); + }); + + it("reusing the same key for a materially different order is a 409, not a silent merge", async () => { + const key = `idem_${crypto.randomUUID()}`; + const r1 = await ordersPOST(postReq(BODY, { "Idempotency-Key": key })); + expect(r1.status).toBe(201); + const r2 = await ordersPOST(postReq(BODY_DIFFERENT, { "Idempotency-Key": key })); + expect(r2.status).toBe(409); + const b2 = await r2.json(); + expect(b2.error.code).toBe("idempotency_key_conflict"); + }); + + it("header name is case-insensitive (Fetch/Next Headers normalize casing)", async () => { + const key = `idem_${crypto.randomUUID()}`; + const r1 = await ordersPOST(postReq(BODY, { "idempotency-key": key })); + const r2 = await ordersPOST(postReq(BODY, { "IDEMPOTENCY-KEY": key })); + expect(r1.status).toBe(201); + expect(r2.status).toBe(200); + }); +}); diff --git a/tests/auth/internal-token.test.ts b/tests/auth/internal-token.test.ts new file mode 100644 index 0000000..43c9860 --- /dev/null +++ b/tests/auth/internal-token.test.ts @@ -0,0 +1,53 @@ +import { describe, it, expect } from "vitest"; +import { verifyInternalServiceToken, INTERNAL_TOKEN_HEADER } from "@/src/auth/internal-token"; + +function headers(map: Record = {}): { get(name: string): string | null } { + const lower = new Map(Object.entries(map).map(([k, v]) => [k.toLowerCase(), v])); + return { get: (name: string) => lower.get(name.toLowerCase()) ?? null }; +} + +describe("verifyInternalServiceToken", () => { + it("accepts a header that matches the configured INTERNAL_TOKEN exactly", () => { + const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: "s3cr3t" }), { + INTERNAL_TOKEN: "s3cr3t", + }); + expect(ok).toBe(true); + }); + + it("rejects a mismatched token", () => { + const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: "wrong" }), { + INTERNAL_TOKEN: "s3cr3t", + }); + expect(ok).toBe(false); + }); + + it("rejects a missing header even when INTERNAL_TOKEN is configured", () => { + const ok = verifyInternalServiceToken(headers({}), { INTERNAL_TOKEN: "s3cr3t" }); + expect(ok).toBe(false); + }); + + it("FAILS CLOSED: an unset INTERNAL_TOKEN never authenticates, even against an empty header value", () => { + // Regression guard: quality_bridge.py sends `os.getenv("INTERNAL_TOKEN", "")` + // -- if the edge's own INTERNAL_TOKEN were also unset, a naive `provided + // === expected` comparison of two empty strings would wrongly authenticate. + const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: "" }), { INTERNAL_TOKEN: undefined }); + expect(ok).toBe(false); + }); + + it("FAILS CLOSED: an unset INTERNAL_TOKEN never authenticates even a non-empty header", () => { + const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: "anything" }), { + INTERNAL_TOKEN: undefined, + }); + expect(ok).toBe(false); + }); + + it("rejects a whitespace-only configured token (trimmed to empty)", () => { + const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: " " }), { INTERNAL_TOKEN: " " }); + expect(ok).toBe(false); + }); + + it("is case-sensitive on the header name lookup contract but tolerant of header-name casing via the HeaderLookup adapter", () => { + const ok = verifyInternalServiceToken(headers({ "X-Internal-Token": "s3cr3t" }), { INTERNAL_TOKEN: "s3cr3t" }); + expect(ok).toBe(true); + }); +}); diff --git a/tests/auth/middleware-internal-token.test.ts b/tests/auth/middleware-internal-token.test.ts new file mode 100644 index 0000000..ec57c4f --- /dev/null +++ b/tests/auth/middleware-internal-token.test.ts @@ -0,0 +1,75 @@ +/** + * P014 regression test: before this pass, POST /api/0pt/quality/bridge-run + * -- the ONE route the Python engine's quality_bridge.py actually calls, + * using X-Internal-Token, never a Bearer token -- was rejected 401 by + * middleware.ts in any environment enforcing auth (the default), because + * middleware required a verified Zitadel Bearer token unconditionally on + * every /api/0pt/* path. See src/auth/internal-token.ts's doc comment for + * the full story. Neither existing test suite caught it: the TS bridge test + * calls the route's POST function directly (bypassing middleware.ts + * entirely) and the Python test mocks the HTTP layer. + * + * This test exercises the actual `middleware` function, not the route + * handler, so it proves the fix at the layer where the bug lived. + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { NextRequest } from 'next/server'; +import { middleware } from '../../middleware'; + +function req(path: string, headers: Record = {}): NextRequest { + return new NextRequest(new URL(path, 'http://localhost'), { headers }); +} + +describe('middleware.ts: internal-token path for /api/0pt/quality/bridge-run', () => { + const saved = { auth: process.env.AUTH_ENFORCE, internal: process.env.INTERNAL_TOKEN, node: process.env.NODE_ENV }; + + beforeEach(() => { + delete process.env.AUTH_ENFORCE; // default posture: enforced + process.env.INTERNAL_TOKEN = 'engine-shared-secret'; + }); + + afterEach(() => { + for (const [k, v] of Object.entries(saved)) { + const envKey = k === 'auth' ? 'AUTH_ENFORCE' : k === 'internal' ? 'INTERNAL_TOKEN' : 'NODE_ENV'; + if (v === undefined) delete process.env[envKey]; + else process.env[envKey] = v; + } + }); + + it('REGRESSION: a valid X-Internal-Token lets the engine through with NO Bearer token (was: 401)', async () => { + const res = await middleware( + req('/api/0pt/quality/bridge-run', { 'x-internal-token': 'engine-shared-secret' }), + ); + // NextResponse.next() carries no body/status override -- it is not an + // error JSON response the way a rejection is. + expect(res.status).toBe(200); + }); + + it('an invalid X-Internal-Token still falls through to the ordinary Bearer requirement (401, not a bypass)', async () => { + const res = await middleware(req('/api/0pt/quality/bridge-run', { 'x-internal-token': 'wrong-secret' })); + expect(res.status).toBe(401); + const body = await res.json(); + expect(body.error).toBe('OPT_UNAUTHENTICATED'); + }); + + it('a valid X-Internal-Token on a DIFFERENT /api/0pt/* route does NOT bypass auth (not a blanket exemption)', async () => { + const res = await middleware(req('/api/0pt/sources', { 'x-internal-token': 'engine-shared-secret' })); + expect(res.status).toBe(401); + }); + + it('no INTERNAL_TOKEN configured on the edge -- the internal-token path never activates, even with the right-looking header', async () => { + delete process.env.INTERNAL_TOKEN; + const res = await middleware( + req('/api/0pt/quality/bridge-run', { 'x-internal-token': 'engine-shared-secret' }), + ); + expect(res.status).toBe(401); + }); + + it('a real Bearer-authenticated caller can still reach bridge-run directly (not internal-token-only)', async () => { + // No X-Internal-Token at all, no Authorization header either -> falls to + // the normal Bearer requirement -> 401, proving this path is additive, + // not a replacement for the existing Zitadel flow. + const res = await middleware(req('/api/0pt/quality/bridge-run')); + expect(res.status).toBe(401); + }); +}); diff --git a/tests/http/tenant-guard.test.ts b/tests/http/tenant-guard.test.ts new file mode 100644 index 0000000..221fc55 --- /dev/null +++ b/tests/http/tenant-guard.test.ts @@ -0,0 +1,48 @@ +import { describe, it, expect } from "vitest"; +import { NextRequest } from "next/server"; +import { assertTenantMatchesVerified, VERIFIED_TENANT_HEADER } from "@/src/http/tenant-guard"; + +function req(headers: Record = {}): NextRequest { + return new NextRequest("http://localhost/api/0pt/sources", { headers }); +} + +describe("assertTenantMatchesVerified", () => { + it("returns 403 when the client-supplied tenant_id disagrees with the verified header", () => { + const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_a" }), "t_b"); + expect(res).not.toBeNull(); + expect(res!.status).toBe(403); + }); + + it("the 403 body names both the requested and verified tenant, without leaking which is 'right'", async () => { + const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_a" }), "t_b"); + const body = await res!.json(); + expect(body.error).toBe("OPT_ACCESS_DENIED"); + expect(body.details).toEqual({ requested_tenant: "t_b", verified_tenant: "t_a" }); + }); + + it("returns null (proceed) when the tenant matches", () => { + const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_a" }), "t_a"); + expect(res).toBeNull(); + }); + + it("returns null (proceed) when no verified-tenant header is present (local-dev escape hatch)", () => { + const res = assertTenantMatchesVerified(req({}), "t_b"); + expect(res).toBeNull(); + }); + + it("returns null when tenantId itself is missing/empty (handlers 400 on that separately)", () => { + const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_a" }), null); + expect(res).toBeNull(); + }); + + it("a spoofed verified-tenant header cannot be used to APPROVE a mismatched request -- only middleware.ts ever sets this header for real", () => { + // This is a documentation-by-test of the trust boundary: the guard trusts + // whatever is in the header, which is exactly why every route sits behind + // middleware.ts (matcher /api/0pt/:path*) that overwrites it with the + // real verified value -- a caller hitting the route directly, outside + // Next's routing, cannot reach the handler without going through + // middleware first in a real deployment. + const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_attacker" }), "t_attacker"); + expect(res).toBeNull(); // matches -- proves the guard is a pure equality check, no extra magic + }); +}); diff --git a/tests/lib/db-ssl-mode.test.ts b/tests/lib/db-ssl-mode.test.ts new file mode 100644 index 0000000..710ad77 --- /dev/null +++ b/tests/lib/db-ssl-mode.test.ts @@ -0,0 +1,39 @@ +/** + * P014: lib/db.ts's `getSql()` used to hardcode `ssl: "require"` + * unconditionally. That is correct for every REAL deployment target this + * app has (Vercel Postgres, managed Postgres) but made ANY plain, + * non-TLS Postgres -- compose.dev.yaml's own local-dev container, or a + * throwaway restore-drill target -- silently unreachable through this + * function, contradicting compose.dev.yaml's own doc comment and its + * "UNVERIFIED... not smoke-tested" caveat. Found and fixed while executing + * P014's real restore drill (see docs/runbook/backup-restore.md). + */ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { sslMode } from "@/lib/db"; + +describe("sslMode", () => { + const saved = process.env.DATABASE_SSL; + afterEach(() => { + if (saved === undefined) delete process.env.DATABASE_SSL; + else process.env.DATABASE_SSL = saved; + }); + beforeEach(() => { + delete process.env.DATABASE_SSL; + }); + + it("defaults to \"require\" when DATABASE_SSL is unset -- unchanged behavior for every real deployment", () => { + expect(sslMode()).toBe("require"); + }); + + for (const value of ["false", "FALSE", "disable", "off", "0", " false "]) { + it(`treats DATABASE_SSL=${JSON.stringify(value)} as an explicit opt-out (false)`, () => { + process.env.DATABASE_SSL = value; + expect(sslMode()).toBe(false); + }); + } + + it("an unrecognized value falls back to the safe default (\"require\"), not an accidental opt-out", () => { + process.env.DATABASE_SSL = "yes-please"; + expect(sslMode()).toBe("require"); + }); +}); diff --git a/tests/lib/orders-idempotency.test.ts b/tests/lib/orders-idempotency.test.ts new file mode 100644 index 0000000..47382b9 --- /dev/null +++ b/tests/lib/orders-idempotency.test.ts @@ -0,0 +1,84 @@ +/** + * P014: lib/orders-store.ts previously had ZERO test coverage and no + * idempotency mechanism -- a retried POST /api/orders created a second, + * indistinguishable row every time (recon: "Duplicate delivery idempotent: + * NOT MET"). These tests exercise the in-memory fallback path directly + * (no DATABASE_URL is set in this test environment, matching every other + * test in this repo) and prove: a retry under the same Idempotency-Key + * replays the original order; a reused key with a different submission is + * refused as a conflict, not silently merged; and omitting the key + * preserves the exact pre-P014 "always create a new order" behavior. + */ +import { describe, it, expect, beforeEach } from "vitest"; +import { + createOrder, + findOrderByIdempotencyKey, + IdempotencyKeyConflictError, + type OrderConfig, +} from "@/lib/orders-store"; + +const CONFIG_A: OrderConfig = { markets: ["SE"], type: "Bolag", fieldIds: ["identity"], volume: 1000 }; +const CONFIG_B: OrderConfig = { markets: ["NO"], type: "Bolag", fieldIds: ["identity"], volume: 5000 }; + +describe("createOrder idempotency (in-memory fallback)", () => { + it("no Idempotency-Key: every call creates a distinct order (unchanged pre-P014 behavior)", async () => { + const first = await createOrder("custom", CONFIG_A); + const second = await createOrder("custom", CONFIG_A); + expect(first.order.order_id).not.toBe(second.order.order_id); + expect(first.replayed).toBe(false); + expect(second.replayed).toBe(false); + }); + + it("same Idempotency-Key + same (flow, config): the second call replays the first order, no duplicate row", async () => { + const key = `idem_${crypto.randomUUID()}`; + const first = await createOrder("custom", CONFIG_A, key); + const second = await createOrder("custom", CONFIG_A, key); + + expect(first.replayed).toBe(false); + expect(second.replayed).toBe(true); + expect(second.order.order_id).toBe(first.order.order_id); + expect(second.order.created_at).toBe(first.order.created_at); + }); + + it("retrying the SAME key three times still yields exactly one underlying order", async () => { + const key = `idem_${crypto.randomUUID()}`; + const a = await createOrder("custom", CONFIG_A, key); + const b = await createOrder("custom", CONFIG_A, key); + const c = await createOrder("custom", CONFIG_A, key); + const ids = new Set([a.order.order_id, b.order.order_id, c.order.order_id]); + expect(ids.size).toBe(1); + }); + + it("same Idempotency-Key reused with a DIFFERENT config: refused as a conflict, not silently merged", async () => { + const key = `idem_${crypto.randomUUID()}`; + await createOrder("custom", CONFIG_A, key); + await expect(createOrder("custom", CONFIG_B, key)).rejects.toBeInstanceOf(IdempotencyKeyConflictError); + }); + + it("same Idempotency-Key reused with a DIFFERENT flow: refused as a conflict", async () => { + const key = `idem_${crypto.randomUUID()}`; + await createOrder("custom", CONFIG_A, key); + await expect(createOrder("subscription", CONFIG_A, key)).rejects.toBeInstanceOf(IdempotencyKeyConflictError); + }); + + it("a conflicting retry does not mutate or replace the original order", async () => { + const key = `idem_${crypto.randomUUID()}`; + const original = await createOrder("custom", CONFIG_A, key); + await expect(createOrder("custom", CONFIG_B, key)).rejects.toThrow(); + const stillThere = await findOrderByIdempotencyKey(key); + expect(stillThere?.order_id).toBe(original.order.order_id); + expect(stillThere?.config).toEqual(CONFIG_A); + }); + + it("different Idempotency-Keys for the same config create independent orders", async () => { + const a = await createOrder("custom", CONFIG_A, `idem_${crypto.randomUUID()}`); + const b = await createOrder("custom", CONFIG_A, `idem_${crypto.randomUUID()}`); + expect(a.order.order_id).not.toBe(b.order.order_id); + }); + + it("an empty-string Idempotency-Key is treated as absent (no dedup, no crash)", async () => { + const a = await createOrder("custom", CONFIG_A, ""); + const b = await createOrder("custom", CONFIG_A, " "); + expect(a.order.order_id).not.toBe(b.order.order_id); + }); +}); diff --git a/tests/validation/0pt-schemas.test.ts b/tests/validation/0pt-schemas.test.ts index b4e2160..31e64e2 100644 --- a/tests/validation/0pt-schemas.test.ts +++ b/tests/validation/0pt-schemas.test.ts @@ -127,6 +127,11 @@ describe('mapDatasetToIcebergTableSchema', () => { describe('requestIcebergWriteSchema', () => { const valid = { + // P014: tenant_id is now required -- the route cross-checks it against + // the caller's verified tenant and threads it into + // IcebergWriteService.requestIcebergWrite so a foreign mapping_id can no + // longer be written to (see getMapping's doc comment). + tenant_id: 't1', actor_id: 'a', correlation_id: 'corr', schema_version: 'v1', @@ -139,4 +144,8 @@ describe('requestIcebergWriteSchema', () => { const { manifest_uri: _omit, ...bad } = valid; expect(parseBody(requestIcebergWriteSchema, bad).ok).toBe(false); }); + it('rejects a write request missing tenant_id', () => { + const { tenant_id: _omit, ...bad } = valid; + expect(parseBody(requestIcebergWriteSchema, bad).ok).toBe(false); + }); });