From cf37a66d350a40d6029258b6495780b1ae52a897 Mon Sep 17 00:00:00 2001 From: sax3l Date: Mon, 2 Nov 2026 03:48:13 +0100 Subject: [PATCH] feat: persist scoped document upload attempts --- sql/018_document_upload_attempts.sql | 23 +++++ .../upload-attempt-store.ts | 83 +++++++++++++++++++ .../upload-attempt-store.realinfra.test.ts | 60 ++++++++++++++ 3 files changed, 166 insertions(+) create mode 100644 sql/018_document_upload_attempts.sql create mode 100644 src/document-workbench/upload-attempt-store.ts create mode 100644 tests/document-workbench/upload-attempt-store.realinfra.test.ts diff --git a/sql/018_document_upload_attempts.sql b/sql/018_document_upload_attempts.sql new file mode 100644 index 0000000..e1b6307 --- /dev/null +++ b/sql/018_document_upload_attempts.sql @@ -0,0 +1,23 @@ +-- Internal coordination only: raw bytes remain in the document storage service. +CREATE TABLE opt_document_upload_attempts ( + tenant_id text NOT NULL CHECK (length(tenant_id) BETWEEN 1 AND 256), + project_id text NOT NULL CHECK (length(project_id) BETWEEN 1 AND 256), + actor_id text NOT NULL CHECK (length(actor_id) BETWEEN 1 AND 256), + attempt_id text NOT NULL CHECK (length(attempt_id) BETWEEN 1 AND 256), + source_sha256 text NOT NULL CHECK (source_sha256 ~ '^[0-9a-f]{64}$'), + source_size_bytes bigint NOT NULL CHECK (source_size_bytes > 0 AND source_size_bytes <= 9007199254740991), + request_fingerprint text NOT NULL CHECK (request_fingerprint ~ '^[0-9a-f]{64}$'), + state text NOT NULL CHECK (state IN ('prepared', 'dispatch_started')), + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + updated_at timestamptz NOT NULL DEFAULT clock_timestamp(), + PRIMARY KEY (tenant_id, attempt_id) +); +ALTER TABLE opt_document_upload_attempts ENABLE ROW LEVEL SECURITY; +ALTER TABLE opt_document_upload_attempts FORCE ROW LEVEL SECURITY; +CREATE POLICY opt_document_upload_attempt_scope ON opt_document_upload_attempts + USING (tenant_id = current_setting('app.opt_tenant_id', true) + AND project_id = current_setting('app.opt_project_id', true) + AND actor_id = current_setting('app.opt_actor_id', true)) + WITH CHECK (tenant_id = current_setting('app.opt_tenant_id', true) + AND project_id = current_setting('app.opt_project_id', true) + AND actor_id = current_setting('app.opt_actor_id', true)); diff --git a/src/document-workbench/upload-attempt-store.ts b/src/document-workbench/upload-attempt-store.ts new file mode 100644 index 0000000..0fa0aea --- /dev/null +++ b/src/document-workbench/upload-attempt-store.ts @@ -0,0 +1,83 @@ +import type { RepositorySql } from '../repositories/db'; +import type postgres from 'postgres'; + +/** All scope values must come from authenticated and project-authorized context. */ +export interface UploadAttemptIdentity { + tenantId: string; + projectId: string; + actorId: string; + /** Also used as the original DOC0 idempotency key by the future caller. */ + attemptId: string; + sourceSha256: string; + sourceSizeBytes: number; + requestFingerprint: string; +} +export type UploadAttemptState = 'prepared' | 'dispatch_started'; +interface Row { + source_sha256: string; + source_size_bytes: string; + request_fingerprint: string; + state: UploadAttemptState; +} + +function validate(input: UploadAttemptIdentity): void { + if (![input.tenantId, input.projectId, input.actorId, input.attemptId].every( + (value) => typeof value === 'string' && value.trim().length > 0 && value.length <= 256, + ) || !/^[0-9a-f]{64}$/.test(input.sourceSha256) || !/^[0-9a-f]{64}$/.test(input.requestFingerprint) + || !Number.isSafeInteger(input.sourceSizeBytes) || input.sourceSizeBytes <= 0) { + throw new Error('Invalid document upload attempt'); + } +} + +/** + * Internal, one-way coordination primitive; not an authentication, project + * authorization or content-integrity check. Callers must persist the exact + * DOC0-compatible fingerprint and verify actual uploaded bytes themselves. + * RLS session settings are set only inside each transaction; the database + * role must not be superuser or BYPASSRLS. A started dispatch is never + * automatically retried or declared absent. No customer route is wired. + */ +export class PostgresUploadAttemptStore { + constructor(private readonly sql: RepositorySql) {} + + private async scoped(input: UploadAttemptIdentity, action: (tx: postgres.TransactionSql) => Promise): Promise { + validate(input); + return this.sql.begin(async (tx) => { + const roles = await tx<{ rolsuper: boolean; rolbypassrls: boolean }[]>` + SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user`; + if (!roles[0] || roles[0].rolsuper || roles[0].rolbypassrls) throw new Error('Upload attempts require an RLS-enforced database role'); + await tx`SELECT set_config('app.opt_tenant_id', ${input.tenantId}, true), set_config('app.opt_project_id', ${input.projectId}, true), set_config('app.opt_actor_id', ${input.actorId}, true)`; + return action(tx); + }) as Promise; + } + + async prepare(input: UploadAttemptIdentity): Promise<{ state: UploadAttemptState; replay: boolean }> { + return this.scoped(input, async (tx) => { + const inserted = await tx` + INSERT INTO opt_document_upload_attempts (tenant_id,project_id,actor_id,attempt_id,source_sha256,source_size_bytes,request_fingerprint,state) + VALUES (${input.tenantId},${input.projectId},${input.actorId},${input.attemptId},${input.sourceSha256},${input.sourceSizeBytes},${input.requestFingerprint},'prepared') + ON CONFLICT (tenant_id,attempt_id) DO NOTHING RETURNING source_sha256,source_size_bytes,request_fingerprint,state`; + const rows = inserted.length ? inserted : await tx` + SELECT source_sha256,source_size_bytes,request_fingerprint,state FROM opt_document_upload_attempts + WHERE tenant_id=${input.tenantId} AND project_id=${input.projectId} AND actor_id=${input.actorId} AND attempt_id=${input.attemptId}`; + const row = rows[0]; + if (!row || row.source_sha256 !== input.sourceSha256 || row.source_size_bytes !== String(input.sourceSizeBytes) || row.request_fingerprint !== input.requestFingerprint) { + throw new Error('Document upload attempt cannot be reused'); + } + return { state: row.state, replay: inserted.length === 0 }; + }); + } + + /** Exactly one claimant may start dispatch. Persist this before any network write. */ + async startDispatch(input: UploadAttemptIdentity): Promise { + return this.scoped(input, async (tx) => { + const rows = await tx<{ attempt_id: string }[]>` + UPDATE opt_document_upload_attempts SET state='dispatch_started', updated_at=clock_timestamp() + WHERE tenant_id=${input.tenantId} AND project_id=${input.projectId} AND actor_id=${input.actorId} + AND attempt_id=${input.attemptId} AND source_sha256=${input.sourceSha256} + AND source_size_bytes=${input.sourceSizeBytes} AND request_fingerprint=${input.requestFingerprint} + AND state='prepared' RETURNING attempt_id`; + return rows.length === 1; + }); + } +} diff --git a/tests/document-workbench/upload-attempt-store.realinfra.test.ts b/tests/document-workbench/upload-attempt-store.realinfra.test.ts new file mode 100644 index 0000000..a17b61c --- /dev/null +++ b/tests/document-workbench/upload-attempt-store.realinfra.test.ts @@ -0,0 +1,60 @@ +import { readFile } from 'node:fs/promises'; +import postgres from 'postgres'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { PostgresUploadAttemptStore, type UploadAttemptIdentity } from '../../src/document-workbench/upload-attempt-store'; + +const url = process.env.OPT_ATTEMPT_TEST_DATABASE_URL; +describe.skipIf(!url)('upload attempt manifest with real PostgreSQL RLS', () => { + let admin: ReturnType; + let app: ReturnType; + let store: PostgresUploadAttemptStore; + let scopedUrl: string; + const sample: UploadAttemptIdentity = { + tenantId: 'tenant-a', projectId: 'project-a', actorId: 'actor-a', attemptId: 'attempt-a', + sourceSha256: 'a'.repeat(64), sourceSizeBytes: 12, requestFingerprint: 'b'.repeat(64), + }; + + beforeAll(async () => { + if (process.env.OPT_ATTEMPT_TEST_ALLOW_LOCAL !== '1' || new URL(url!).hostname !== '127.0.0.1') { + throw new Error('Only an explicitly enabled isolated loopback test database may be used'); + } + admin = postgres(url!, { max: 1, prepare: false }); + await admin.unsafe(await readFile('sql/018_document_upload_attempts.sql', 'utf8')); + await admin.unsafe('CREATE ROLE opt_attempt_test LOGIN NOSUPERUSER NOBYPASSRLS'); + await admin.unsafe('GRANT USAGE ON SCHEMA public TO opt_attempt_test'); + await admin.unsafe('GRANT SELECT, INSERT, UPDATE ON opt_document_upload_attempts TO opt_attempt_test'); + const appUrl = new URL(url!); + appUrl.username = 'opt_attempt_test'; + scopedUrl = appUrl.toString(); + app = postgres(scopedUrl, { max: 8, prepare: false }); + store = new PostgresUploadAttemptStore(app); + }); + afterAll(async () => { await app?.end(); await admin?.end(); }); + + it('rejects a database role which bypasses RLS', async () => { + await expect(new PostgresUploadAttemptStore(admin).prepare(sample)).rejects.toThrow(/RLS-enforced/); + }); + it('persists matching retries and rejects changed or differently scoped identities', async () => { + expect(await store.prepare(sample)).toEqual({ state: 'prepared', replay: false }); + expect(await store.prepare(sample)).toEqual({ state: 'prepared', replay: true }); + await expect(store.prepare({ ...sample, sourceSha256: 'c'.repeat(64) })).rejects.toThrow(); + await expect(store.prepare({ ...sample, projectId: 'project-b' })).rejects.toThrow('Document upload attempt cannot be reused'); + await expect(store.prepare({ ...sample, actorId: 'actor-b' })).rejects.toThrow('Document upload attempt cannot be reused'); + expect(await store.prepare({ ...sample, tenantId: 'tenant-b' })).toEqual({ state: 'prepared', replay: false }); + expect(await app`SELECT * FROM opt_document_upload_attempts`).toHaveLength(0); + }); + it('allows only one concurrent dispatch claimant and survives a new connection', async () => { + const attempt = { ...sample, attemptId: 'race' }; + expect(await store.startDispatch(attempt)).toBe(false); + await store.prepare(attempt); + expect(await store.startDispatch({ ...attempt, actorId: 'actor-b' })).toBe(false); + expect(await store.startDispatch({ ...attempt, sourceSha256: 'c'.repeat(64) })).toBe(false); + const results = await Promise.all(Array.from({ length: 8 }, () => store.startDispatch(attempt))); + expect(results.filter(Boolean)).toHaveLength(1); + expect(await store.startDispatch(attempt)).toBe(false); + expect(await store.prepare(attempt)).toEqual({ state: 'dispatch_started', replay: true }); + const second = postgres(scopedUrl, { max: 1, prepare: false }); + try { expect(await new PostgresUploadAttemptStore(second).prepare(attempt)).toEqual({ state: 'dispatch_started', replay: true }); } + finally { await second.end(); } + }); +}); -- 2.54.0