From ec956ff6a65d8dcd18f56b3abdcd63c7351504b0 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 28 Oct 2026 08:37:36 -0700 Subject: [PATCH 1/6] feat(storage): partner file metadata + 30-day retention schema (v138) Add partner_files / partner_file_derived / partner_file_deletion_log with CASE RLS, trigger-maintained delete_after (+30d), and private partner-uploads bucket policies. Mirror on Alembic; nullable raw_bytes + payload check on fsma.manifest_blobs for later hash/pointer cutover. Migration-only; do not apply without C.Lee yes. Synthetic fixtures only until SEL-590. Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- ...901_v138_partner_file_storage_retention.py | 369 ++++++++++++++++++ ...100000_partner_files_storage_retention.sql | 256 ++++++++++++ .../20261028100010_partner_uploads_bucket.sql | 58 +++ 3 files changed, 683 insertions(+) create mode 100644 alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py create mode 100644 supabase/migrations/20261028100000_partner_files_storage_retention.sql create mode 100644 supabase/migrations/20261028100010_partner_uploads_bucket.sql diff --git a/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py b/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py new file mode 100644 index 000000000..bf0a7887a --- /dev/null +++ b/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py @@ -0,0 +1,369 @@ +"""Partner file storage metadata + 30-day retention (v138). + +Mirrors supabase/migrations/20261028100000_partner_files_storage_retention.sql +exactly for partner_files / partner_file_derived / partner_file_deletion_log +(tables, indexes, CASE RLS, grants). Bucket creation lives only in +supabase/migrations/20261028100010_partner_uploads_bucket.sql (storage schema +is Supabase-only). + +Also prepares Railway fsma.manifest_blobs for hash/pointer cutover: raw_bytes +becomes nullable with a check that at least one of raw_bytes or storage_url is +present (storage_url already exists from v095). Does not move bytes; does not +apply envelope encryption. + +Additive and idempotent. Downgrade drops only what this revision created and +restores raw_bytes NOT NULL when safe. + +Revision ID: b8c1e4f2a901 +Revises: a3f7ce361b0e +Create Date: 2026-10-28 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +revision: str = "b8c1e4f2a901" # pragma: allowlist secret +down_revision: Union[str, Sequence[str], None] = "a3f7ce361b0e" # pragma: allowlist secret +branch_labels = None +depends_on = None + +# Mirrored from the supabase SQL migration (minus the hand-only downgrade notes). +_PARTNER_FILES_SQL = """ +-- Partner file storage metadata + 30-day retention clock (SEL-549 / first-file). +-- +-- Metadata only. Partner file bytes live in the private Supabase Storage bucket +-- `partner-uploads` (see sibling migration 20261028100010). Never store partner +-- file contents in these tables or in the deletion log. +-- +-- Clock: delete_after = report_delivered_at + 30 days (trigger-maintained; UTC). +-- Envelope encryption columns (kek_key_id / wrapped_dek) are nullable; v1 may +-- rely on provider at-rest encryption only (design open Q1 deferred). +-- +-- RLS uses CASE + (select public.get_tenant_context()), matching epcis / +-- data_gap (not the JWT sketch). Idempotent. Do not apply to prod without +-- C.Lee yes. Synthetic fixtures only until SEL-590 clears. + +CREATE TABLE IF NOT EXISTS public.partner_files ( + id uuid primary key default gen_random_uuid(), + tenant_id uuid not null references public.tenants(id), + engagement_id uuid not null, + storage_bucket text not null default 'partner-uploads', + storage_path text not null, + content_sha256 text not null, + ciphertext_sha256 text, + byte_size bigint not null, + content_type text not null, + original_filename text not null, + kek_key_id text, + wrapped_dek bytea, + uploaded_at timestamptz not null default now(), + report_delivered_at timestamptz, + -- Maintained by partner_files_set_delete_after trigger (timestamptz + interval + -- is not IMMUTABLE, so a generated column is not allowed). + delete_after timestamptz, + deleted_at timestamptz, + status text not null default 'uploaded', + constraint partner_files_byte_size_check + check (byte_size > 0), + constraint partner_files_content_sha256_check + check (content_sha256 ~ '^[0-9a-f]{64}$'), + constraint partner_files_ciphertext_sha256_check + check (ciphertext_sha256 is null or ciphertext_sha256 ~ '^[0-9a-f]{64}$'), + constraint partner_files_status_check + check (status in ( + 'uploaded', + 'processing', + 'report_ready', + 'delivered', + 'pending_delete', + 'deleted' + )), + constraint partner_files_storage_bucket_check + check (storage_bucket = 'partner-uploads'), + constraint partner_files_storage_path_unique + unique (storage_bucket, storage_path) +); + +CREATE INDEX IF NOT EXISTS partner_files_due_delete_idx + ON public.partner_files (delete_after) + WHERE deleted_at IS NULL AND report_delivered_at IS NOT NULL; + +CREATE INDEX IF NOT EXISTS partner_files_tenant_uploaded_idx + ON public.partner_files (tenant_id, uploaded_at desc); + +-- delete_after = report_delivered_at + 30 days (trigger; generated col not allowed) +CREATE OR REPLACE FUNCTION public.partner_files_set_delete_after() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +BEGIN + IF NEW.report_delivered_at IS NULL THEN + NEW.delete_after := NULL; + ELSE + NEW.delete_after := NEW.report_delivered_at + interval '30 days'; + END IF; + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS partner_files_set_delete_after ON public.partner_files; +CREATE TRIGGER partner_files_set_delete_after + BEFORE INSERT OR UPDATE OF report_delivered_at + ON public.partner_files + FOR EACH ROW + EXECUTE FUNCTION public.partner_files_set_delete_after(); + +CREATE TABLE IF NOT EXISTS public.partner_file_derived ( + id uuid primary key default gen_random_uuid(), + partner_file_id uuid not null references public.partner_files(id) on delete cascade, + tenant_id uuid not null references public.tenants(id), + kind text not null, + storage_bucket text, + storage_path text, + row_ref text, + created_at timestamptz not null default now(), + deleted_at timestamptz, + constraint partner_file_derived_kind_check + check (char_length(kind) between 1 and 64) +); + +CREATE INDEX IF NOT EXISTS partner_file_derived_file_idx + ON public.partner_file_derived (partner_file_id) + WHERE deleted_at IS NULL; + +CREATE TABLE IF NOT EXISTS public.partner_file_deletion_log ( + id uuid primary key default gen_random_uuid(), + partner_file_id uuid not null, + tenant_id uuid not null, + engagement_id uuid, + storage_bucket text not null, + storage_path text not null, + content_sha256 text not null, + ciphertext_sha256 text, + byte_size bigint, + report_delivered_at timestamptz, + deleted_at timestamptz not null default now(), + actor text not null, + reason text not null, + entry_hash text not null, + prev_entry_hash text, + constraint partner_file_deletion_log_content_sha256_check + check (content_sha256 ~ '^[0-9a-f]{64}$'), + constraint partner_file_deletion_log_ciphertext_sha256_check + check (ciphertext_sha256 is null or ciphertext_sha256 ~ '^[0-9a-f]{64}$'), + constraint partner_file_deletion_log_entry_hash_check + check (entry_hash ~ '^[0-9a-f]{64}$'), + constraint partner_file_deletion_log_prev_entry_hash_check + check (prev_entry_hash is null or prev_entry_hash ~ '^[0-9a-f]{64}$'), + constraint partner_file_deletion_log_actor_check + check ( + actor = 'retention_job' + or actor = 'partner_request' + or actor like 'ops:%' + ), + constraint partner_file_deletion_log_reason_check + check (reason in ('retention_30d', 'partner_request_15d', 'terms_end')) +); + +CREATE INDEX IF NOT EXISTS partner_file_deletion_log_tenant_deleted_idx + ON public.partner_file_deletion_log (tenant_id, deleted_at desc); + +ALTER TABLE public.partner_files ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.partner_files FORCE ROW LEVEL SECURITY; +ALTER TABLE public.partner_file_derived ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.partner_file_derived FORCE ROW LEVEL SECURITY; +ALTER TABLE public.partner_file_deletion_log ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.partner_file_deletion_log FORCE ROW LEVEL SECURITY; + +-- CASE form, not OR: public.get_tenant_context() raises when app.tenant_id is +-- unset and Postgres does not order OR operands, so the sysadmin arm is tested +-- first (same predicate shape as alembic v096 / v134 / v135 / v137). +DROP POLICY IF EXISTS tenant_isolation_partner_files ON public.partner_files; +CREATE POLICY tenant_isolation_partner_files ON public.partner_files + FOR ALL + USING ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ) + WITH CHECK ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ); + +DROP POLICY IF EXISTS tenant_isolation_partner_file_derived ON public.partner_file_derived; +CREATE POLICY tenant_isolation_partner_file_derived ON public.partner_file_derived + FOR ALL + USING ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ) + WITH CHECK ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ); + +DROP POLICY IF EXISTS tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log; +CREATE POLICY tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log + FOR SELECT + USING ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ); + +-- Inserts into the deletion log go through the retention worker (regengine / +-- SECURITY DEFINER RPC). No authenticated UPDATE/DELETE on the log. +DROP POLICY IF EXISTS partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log; +CREATE POLICY partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log + FOR INSERT + WITH CHECK ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ); + +DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'anon') THEN + REVOKE ALL ON public.partner_files FROM anon; + REVOKE ALL ON public.partner_file_derived FROM anon; + REVOKE ALL ON public.partner_file_deletion_log FROM anon; + END IF; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'authenticated') THEN + -- Metadata reads/writes for tenant users go through the app role / + -- signed upload path, not broad authenticated table DML. + REVOKE ALL ON public.partner_files FROM authenticated; + REVOKE ALL ON public.partner_file_derived FROM authenticated; + REVOKE ALL ON public.partner_file_deletion_log FROM authenticated; + END IF; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine') THEN + GRANT SELECT, INSERT, UPDATE ON public.partner_files TO regengine; + GRANT SELECT, INSERT, UPDATE ON public.partner_file_derived TO regengine; + -- Append-only audit: INSERT + SELECT only. Never UPDATE/DELETE. + GRANT SELECT, INSERT ON public.partner_file_deletion_log TO regengine; + REVOKE UPDATE, DELETE ON public.partner_file_deletion_log FROM regengine; + END IF; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine_sysadmin') THEN + GRANT SELECT ON public.partner_files TO regengine_sysadmin; + GRANT SELECT ON public.partner_file_derived TO regengine_sysadmin; + GRANT SELECT ON public.partner_file_deletion_log TO regengine_sysadmin; + END IF; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'service_role') THEN + -- Prefer SECURITY DEFINER RPCs; keep service_role from mutating the log. + REVOKE UPDATE, DELETE ON public.partner_file_deletion_log FROM service_role; + END IF; +END $$; + +""" + +_MANIFEST_BLOBS_SQL = """ +-- Railway cutover helper: allow hash/pointer-only manifest_blobs rows. +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'fsma' + AND table_name = 'manifest_blobs' + AND column_name = 'raw_bytes' + ) THEN + ALTER TABLE fsma.manifest_blobs ALTER COLUMN raw_bytes DROP NOT NULL; + END IF; +END $$; + +ALTER TABLE fsma.manifest_blobs DROP CONSTRAINT IF EXISTS manifest_blobs_payload_present_check; +ALTER TABLE fsma.manifest_blobs + ADD CONSTRAINT manifest_blobs_payload_present_check + CHECK ( + raw_bytes IS NOT NULL + OR (storage_url IS NOT NULL AND length(btrim(storage_url)) > 0) + ); +""" + + +def upgrade() -> None: + # Split on semicolons outside dollar-quoted blocks so DO $$ ... $$; stays intact. + def _statements(sql: str) -> list[str]: + out: list[str] = [] + buf: list[str] = [] + in_dollar = False + i = 0 + while i < len(sql): + if sql.startswith("$$", i): + in_dollar = not in_dollar + buf.append("$$") + i += 2 + continue + ch = sql[i] + if ch == ";" and not in_dollar: + stmt = "".join(buf).strip() + if stmt and not all( + line.strip().startswith("--") or not line.strip() + for line in stmt.splitlines() + ): + out.append(stmt) + buf = [] + i += 1 + continue + buf.append(ch) + i += 1 + tail = "".join(buf).strip() + if tail: + out.append(tail) + return out + + for statement in _statements(_PARTNER_FILES_SQL) + _statements(_MANIFEST_BLOBS_SQL): + op.execute(sa.text(statement)) + + +def downgrade() -> None: + op.execute( + sa.text( + """ +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'fsma' AND table_name = 'manifest_blobs' + ) THEN + ALTER TABLE fsma.manifest_blobs DROP CONSTRAINT IF EXISTS manifest_blobs_payload_present_check; + IF NOT EXISTS ( + SELECT 1 FROM fsma.manifest_blobs WHERE raw_bytes IS NULL + ) THEN + ALTER TABLE fsma.manifest_blobs ALTER COLUMN raw_bytes SET NOT NULL; + END IF; + END IF; +END $$; +""" + ) + ) + op.execute(sa.text("DROP POLICY IF EXISTS partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log")) + op.execute(sa.text("DROP POLICY IF EXISTS tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log")) + op.execute(sa.text("DROP POLICY IF EXISTS tenant_isolation_partner_file_derived ON public.partner_file_derived")) + op.execute(sa.text("DROP POLICY IF EXISTS tenant_isolation_partner_files ON public.partner_files")) + op.execute(sa.text("DROP TRIGGER IF EXISTS partner_files_set_delete_after ON public.partner_files")) + op.execute(sa.text("DROP FUNCTION IF EXISTS public.partner_files_set_delete_after()")) + op.execute(sa.text("DROP INDEX IF EXISTS public.partner_file_deletion_log_tenant_deleted_idx")) + op.execute(sa.text("DROP INDEX IF EXISTS public.partner_file_derived_file_idx")) + op.execute(sa.text("DROP INDEX IF EXISTS public.partner_files_tenant_uploaded_idx")) + op.execute(sa.text("DROP INDEX IF EXISTS public.partner_files_due_delete_idx")) + op.execute(sa.text("DROP TABLE IF EXISTS public.partner_file_deletion_log")) + op.execute(sa.text("DROP TABLE IF EXISTS public.partner_file_derived")) + op.execute(sa.text("DROP TABLE IF EXISTS public.partner_files")) diff --git a/supabase/migrations/20261028100000_partner_files_storage_retention.sql b/supabase/migrations/20261028100000_partner_files_storage_retention.sql new file mode 100644 index 000000000..6db6e2433 --- /dev/null +++ b/supabase/migrations/20261028100000_partner_files_storage_retention.sql @@ -0,0 +1,256 @@ +-- Partner file storage metadata + 30-day retention clock (SEL-549 / first-file). +-- +-- Metadata only. Partner file bytes live in the private Supabase Storage bucket +-- `partner-uploads` (see sibling migration 20261028100010). Never store partner +-- file contents in these tables or in the deletion log. +-- +-- Clock: delete_after = report_delivered_at + 30 days (trigger-maintained; UTC). +-- Envelope encryption columns (kek_key_id / wrapped_dek) are nullable; v1 may +-- rely on provider at-rest encryption only (design open Q1 deferred). +-- +-- RLS uses CASE + (select public.get_tenant_context()), matching epcis / +-- data_gap (not the JWT sketch). Idempotent. Do not apply to prod without +-- C.Lee yes. Synthetic fixtures only until SEL-590 clears. + +CREATE TABLE IF NOT EXISTS public.partner_files ( + id uuid primary key default gen_random_uuid(), + tenant_id uuid not null references public.tenants(id), + engagement_id uuid not null, + storage_bucket text not null default 'partner-uploads', + storage_path text not null, + content_sha256 text not null, + ciphertext_sha256 text, + byte_size bigint not null, + content_type text not null, + original_filename text not null, + kek_key_id text, + wrapped_dek bytea, + uploaded_at timestamptz not null default now(), + report_delivered_at timestamptz, + -- Maintained by partner_files_set_delete_after trigger (timestamptz + interval + -- is not IMMUTABLE, so a generated column is not allowed). + delete_after timestamptz, + deleted_at timestamptz, + status text not null default 'uploaded', + constraint partner_files_byte_size_check + check (byte_size > 0), + constraint partner_files_content_sha256_check + check (content_sha256 ~ '^[0-9a-f]{64}$'), + constraint partner_files_ciphertext_sha256_check + check (ciphertext_sha256 is null or ciphertext_sha256 ~ '^[0-9a-f]{64}$'), + constraint partner_files_status_check + check (status in ( + 'uploaded', + 'processing', + 'report_ready', + 'delivered', + 'pending_delete', + 'deleted' + )), + constraint partner_files_storage_bucket_check + check (storage_bucket = 'partner-uploads'), + constraint partner_files_storage_path_unique + unique (storage_bucket, storage_path) +); + +CREATE INDEX IF NOT EXISTS partner_files_due_delete_idx + ON public.partner_files (delete_after) + WHERE deleted_at IS NULL AND report_delivered_at IS NOT NULL; + +CREATE INDEX IF NOT EXISTS partner_files_tenant_uploaded_idx + ON public.partner_files (tenant_id, uploaded_at desc); + +-- delete_after = report_delivered_at + 30 days (trigger; generated col not allowed) +CREATE OR REPLACE FUNCTION public.partner_files_set_delete_after() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +BEGIN + IF NEW.report_delivered_at IS NULL THEN + NEW.delete_after := NULL; + ELSE + NEW.delete_after := NEW.report_delivered_at + interval '30 days'; + END IF; + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS partner_files_set_delete_after ON public.partner_files; +CREATE TRIGGER partner_files_set_delete_after + BEFORE INSERT OR UPDATE OF report_delivered_at + ON public.partner_files + FOR EACH ROW + EXECUTE FUNCTION public.partner_files_set_delete_after(); + +CREATE TABLE IF NOT EXISTS public.partner_file_derived ( + id uuid primary key default gen_random_uuid(), + partner_file_id uuid not null references public.partner_files(id) on delete cascade, + tenant_id uuid not null references public.tenants(id), + kind text not null, + storage_bucket text, + storage_path text, + row_ref text, + created_at timestamptz not null default now(), + deleted_at timestamptz, + constraint partner_file_derived_kind_check + check (char_length(kind) between 1 and 64) +); + +CREATE INDEX IF NOT EXISTS partner_file_derived_file_idx + ON public.partner_file_derived (partner_file_id) + WHERE deleted_at IS NULL; + +CREATE TABLE IF NOT EXISTS public.partner_file_deletion_log ( + id uuid primary key default gen_random_uuid(), + partner_file_id uuid not null, + tenant_id uuid not null, + engagement_id uuid, + storage_bucket text not null, + storage_path text not null, + content_sha256 text not null, + ciphertext_sha256 text, + byte_size bigint, + report_delivered_at timestamptz, + deleted_at timestamptz not null default now(), + actor text not null, + reason text not null, + entry_hash text not null, + prev_entry_hash text, + constraint partner_file_deletion_log_content_sha256_check + check (content_sha256 ~ '^[0-9a-f]{64}$'), + constraint partner_file_deletion_log_ciphertext_sha256_check + check (ciphertext_sha256 is null or ciphertext_sha256 ~ '^[0-9a-f]{64}$'), + constraint partner_file_deletion_log_entry_hash_check + check (entry_hash ~ '^[0-9a-f]{64}$'), + constraint partner_file_deletion_log_prev_entry_hash_check + check (prev_entry_hash is null or prev_entry_hash ~ '^[0-9a-f]{64}$'), + constraint partner_file_deletion_log_actor_check + check ( + actor = 'retention_job' + or actor = 'partner_request' + or actor like 'ops:%' + ), + constraint partner_file_deletion_log_reason_check + check (reason in ('retention_30d', 'partner_request_15d', 'terms_end')) +); + +CREATE INDEX IF NOT EXISTS partner_file_deletion_log_tenant_deleted_idx + ON public.partner_file_deletion_log (tenant_id, deleted_at desc); + +ALTER TABLE public.partner_files ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.partner_files FORCE ROW LEVEL SECURITY; +ALTER TABLE public.partner_file_derived ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.partner_file_derived FORCE ROW LEVEL SECURITY; +ALTER TABLE public.partner_file_deletion_log ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.partner_file_deletion_log FORCE ROW LEVEL SECURITY; + +-- CASE form, not OR: public.get_tenant_context() raises when app.tenant_id is +-- unset and Postgres does not order OR operands, so the sysadmin arm is tested +-- first (same predicate shape as alembic v096 / v134 / v135 / v137). +DROP POLICY IF EXISTS tenant_isolation_partner_files ON public.partner_files; +CREATE POLICY tenant_isolation_partner_files ON public.partner_files + FOR ALL + USING ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ) + WITH CHECK ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ); + +DROP POLICY IF EXISTS tenant_isolation_partner_file_derived ON public.partner_file_derived; +CREATE POLICY tenant_isolation_partner_file_derived ON public.partner_file_derived + FOR ALL + USING ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ) + WITH CHECK ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ); + +DROP POLICY IF EXISTS tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log; +CREATE POLICY tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log + FOR SELECT + USING ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ); + +-- Inserts into the deletion log go through the retention worker (regengine / +-- SECURITY DEFINER RPC). No authenticated UPDATE/DELETE on the log. +DROP POLICY IF EXISTS partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log; +CREATE POLICY partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log + FOR INSERT + WITH CHECK ( + CASE + WHEN (select current_setting('regengine.is_sysadmin', true)) = 'true' + AND current_user = 'regengine_sysadmin' THEN true + ELSE tenant_id = (select public.get_tenant_context()) + END + ); + +DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'anon') THEN + REVOKE ALL ON public.partner_files FROM anon; + REVOKE ALL ON public.partner_file_derived FROM anon; + REVOKE ALL ON public.partner_file_deletion_log FROM anon; + END IF; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'authenticated') THEN + -- Metadata reads/writes for tenant users go through the app role / + -- signed upload path, not broad authenticated table DML. + REVOKE ALL ON public.partner_files FROM authenticated; + REVOKE ALL ON public.partner_file_derived FROM authenticated; + REVOKE ALL ON public.partner_file_deletion_log FROM authenticated; + END IF; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine') THEN + GRANT SELECT, INSERT, UPDATE ON public.partner_files TO regengine; + GRANT SELECT, INSERT, UPDATE ON public.partner_file_derived TO regengine; + -- Append-only audit: INSERT + SELECT only. Never UPDATE/DELETE. + GRANT SELECT, INSERT ON public.partner_file_deletion_log TO regengine; + REVOKE UPDATE, DELETE ON public.partner_file_deletion_log FROM regengine; + END IF; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine_sysadmin') THEN + GRANT SELECT ON public.partner_files TO regengine_sysadmin; + GRANT SELECT ON public.partner_file_derived TO regengine_sysadmin; + GRANT SELECT ON public.partner_file_deletion_log TO regengine_sysadmin; + END IF; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'service_role') THEN + -- Prefer SECURITY DEFINER RPCs; keep service_role from mutating the log. + REVOKE UPDATE, DELETE ON public.partner_file_deletion_log FROM service_role; + END IF; +END $$; + +-- Downgrade (run by hand only): +-- +-- DROP POLICY IF EXISTS partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log; +-- DROP POLICY IF EXISTS tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log; +-- DROP POLICY IF EXISTS tenant_isolation_partner_file_derived ON public.partner_file_derived; +-- DROP POLICY IF EXISTS tenant_isolation_partner_files ON public.partner_files; +-- DROP TRIGGER IF EXISTS partner_files_set_delete_after ON public.partner_files; +-- DROP FUNCTION IF EXISTS public.partner_files_set_delete_after(); +-- DROP INDEX IF EXISTS public.partner_file_deletion_log_tenant_deleted_idx; +-- DROP INDEX IF EXISTS public.partner_file_derived_file_idx; +-- DROP INDEX IF EXISTS public.partner_files_tenant_uploaded_idx; +-- DROP INDEX IF EXISTS public.partner_files_due_delete_idx; +-- DROP TABLE IF EXISTS public.partner_file_deletion_log; +-- DROP TABLE IF EXISTS public.partner_file_derived; +-- DROP TABLE IF EXISTS public.partner_files; diff --git a/supabase/migrations/20261028100010_partner_uploads_bucket.sql b/supabase/migrations/20261028100010_partner_uploads_bucket.sql new file mode 100644 index 000000000..b195fe80f --- /dev/null +++ b/supabase/migrations/20261028100010_partner_uploads_bucket.sql @@ -0,0 +1,58 @@ +-- Private Supabase Storage bucket for partner Beta uploads. +-- +-- Layout: partner-uploads/{tenant_id}/{engagement_id}/{partner_file_id} +-- Private (not public). No anonymous list/read. MIME allowlist CSV-oriented. +-- Client UPDATE/DELETE on storage.objects is denied; retention uses service +-- role / SECURITY DEFINER. Folder[1] must equal live tenant context. +-- +-- Supabase-only (storage schema). Not mirrored on Railway Postgres. +-- Idempotent. Do not apply to prod without C.Lee yes. + +INSERT INTO storage.buckets (id, name, public, file_size_limit, allowed_mime_types) +VALUES ( + 'partner-uploads', + 'partner-uploads', + false, + 26214400, -- 25 MiB Beta cap + ARRAY[ + 'text/csv', + 'application/csv', + 'text/plain', + 'application/vnd.ms-excel' + ]::text[] +) +ON CONFLICT (id) DO UPDATE +SET + public = EXCLUDED.public, + file_size_limit = EXCLUDED.file_size_limit, + allowed_mime_types = EXCLUDED.allowed_mime_types; + +-- storage.objects RLS for this bucket. CASE + get_tenant_context (not JWT). +DROP POLICY IF EXISTS partner_uploads_select_own ON storage.objects; +CREATE POLICY partner_uploads_select_own ON storage.objects + FOR SELECT + TO authenticated + USING ( + bucket_id = 'partner-uploads' + AND (storage.foldername(name))[1] = (select public.get_tenant_context())::text + ); + +DROP POLICY IF EXISTS partner_uploads_insert_own ON storage.objects; +CREATE POLICY partner_uploads_insert_own ON storage.objects + FOR INSERT + TO authenticated + WITH CHECK ( + bucket_id = 'partner-uploads' + AND (storage.foldername(name))[1] = (select public.get_tenant_context())::text + ); + +-- Explicitly no UPDATE/DELETE policies for authenticated on this bucket. +DROP POLICY IF EXISTS partner_uploads_update_own ON storage.objects; +DROP POLICY IF EXISTS partner_uploads_delete_own ON storage.objects; + +-- Downgrade (run by hand only): +-- +-- DROP POLICY IF EXISTS partner_uploads_insert_own ON storage.objects; +-- DROP POLICY IF EXISTS partner_uploads_select_own ON storage.objects; +-- DELETE FROM storage.objects WHERE bucket_id = 'partner-uploads'; +-- DELETE FROM storage.buckets WHERE id = 'partner-uploads'; From 290a5a04aea3c8f18c0a2e843d3c863b0e594a1a Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 28 Oct 2026 08:50:15 -0700 Subject: [PATCH 2/6] fix(storage): v138 alembic statement tuple + always-recompute delete_after Codex FAIL on #3729: drop the runtime semicolon splitter (comment `;` broke upgrade) in favor of an explicit _UPGRADE_STATEMENTS tuple like v137, and fire partner_files_set_delete_after BEFORE INSERT OR UPDATE so a direct UPDATE of delete_after cannot desync the retention clock. Stub PG16 tests cover idempotent apply and the overwrite behavior. Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- ...901_v138_partner_file_storage_retention.py | 228 +++++++-------- ...100000_partner_files_storage_retention.sql | 2 +- ...est_v138_partner_file_storage_retention.py | 269 ++++++++++++++++++ 3 files changed, 387 insertions(+), 112 deletions(-) create mode 100644 tests/migrations/test_v138_partner_file_storage_retention.py diff --git a/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py b/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py index bf0a7887a..01b09b86d 100644 --- a/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py +++ b/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py @@ -1,18 +1,20 @@ -"""Partner file storage metadata + 30-day retention (v138). +"""Partner file storage metadata + 30-day retention (v138 / SEL-549 first-file). Mirrors supabase/migrations/20261028100000_partner_files_storage_retention.sql -exactly for partner_files / partner_file_derived / partner_file_deletion_log -(tables, indexes, CASE RLS, grants). Bucket creation lives only in -supabase/migrations/20261028100010_partner_uploads_bucket.sql (storage schema -is Supabase-only). +(metadata + RLS + delete_after trigger) and the manifest_blobs payload-pointer +changes from that same design. Bucket DDL stays in the Supabase-only sibling +20261028100010_partner_uploads_bucket.sql (storage.buckets is not on Railway). -Also prepares Railway fsma.manifest_blobs for hash/pointer cutover: raw_bytes -becomes nullable with a check that at least one of raw_bytes or storage_url is -present (storage_url already exists from v095). Does not move bytes; does not -apply envelope encryption. +Statements are an explicit tuple (same shape as v137) — no runtime semicolon +splitter. Comments may contain `;`; dollar-quoted function / DO bodies stay +inside a single statement string. -Additive and idempotent. Downgrade drops only what this revision created and -restores raw_bytes NOT NULL when safe. +delete_after is maintained by a BEFORE INSERT OR UPDATE trigger that always +recomputes from report_delivered_at (+ 30 days, or NULL), so a direct UPDATE of +delete_after cannot desync the retention clock. + +Additive and idempotent. Do not apply to prod without C.Lee yes. +Synthetic fixtures only until SEL-590 clears. Revision ID: b8c1e4f2a901 Revises: a3f7ce361b0e @@ -22,15 +24,14 @@ from typing import Sequence, Union from alembic import op -import sqlalchemy as sa revision: str = "b8c1e4f2a901" # pragma: allowlist secret down_revision: Union[str, Sequence[str], None] = "a3f7ce361b0e" # pragma: allowlist secret branch_labels = None depends_on = None -# Mirrored from the supabase SQL migration (minus the hand-only downgrade notes). -_PARTNER_FILES_SQL = """ +_UPGRADE_STATEMENTS: tuple[str, ...] = ( + """ -- Partner file storage metadata + 30-day retention clock (SEL-549 / first-file). -- -- Metadata only. Partner file bytes live in the private Supabase Storage bucket @@ -84,15 +85,18 @@ check (storage_bucket = 'partner-uploads'), constraint partner_files_storage_path_unique unique (storage_bucket, storage_path) -); - +) +""", + """ CREATE INDEX IF NOT EXISTS partner_files_due_delete_idx ON public.partner_files (delete_after) - WHERE deleted_at IS NULL AND report_delivered_at IS NOT NULL; - + WHERE deleted_at IS NULL AND report_delivered_at IS NOT NULL +""", + """ CREATE INDEX IF NOT EXISTS partner_files_tenant_uploaded_idx - ON public.partner_files (tenant_id, uploaded_at desc); - + ON public.partner_files (tenant_id, uploaded_at desc) +""", + """ -- delete_after = report_delivered_at + 30 days (trigger; generated col not allowed) CREATE OR REPLACE FUNCTION public.partner_files_set_delete_after() RETURNS trigger @@ -106,15 +110,19 @@ END IF; RETURN NEW; END; -$$; - -DROP TRIGGER IF EXISTS partner_files_set_delete_after ON public.partner_files; +$$ +""", + """ +DROP TRIGGER IF EXISTS partner_files_set_delete_after ON public.partner_files +""", + """ CREATE TRIGGER partner_files_set_delete_after - BEFORE INSERT OR UPDATE OF report_delivered_at + BEFORE INSERT OR UPDATE ON public.partner_files FOR EACH ROW - EXECUTE FUNCTION public.partner_files_set_delete_after(); - + EXECUTE FUNCTION public.partner_files_set_delete_after() +""", + """ CREATE TABLE IF NOT EXISTS public.partner_file_derived ( id uuid primary key default gen_random_uuid(), partner_file_id uuid not null references public.partner_files(id) on delete cascade, @@ -127,12 +135,14 @@ deleted_at timestamptz, constraint partner_file_derived_kind_check check (char_length(kind) between 1 and 64) -); - +) +""", + """ CREATE INDEX IF NOT EXISTS partner_file_derived_file_idx ON public.partner_file_derived (partner_file_id) - WHERE deleted_at IS NULL; - + WHERE deleted_at IS NULL +""", + """ CREATE TABLE IF NOT EXISTS public.partner_file_deletion_log ( id uuid primary key default gen_random_uuid(), partner_file_id uuid not null, @@ -165,22 +175,37 @@ ), constraint partner_file_deletion_log_reason_check check (reason in ('retention_30d', 'partner_request_15d', 'terms_end')) -); - +) +""", + """ CREATE INDEX IF NOT EXISTS partner_file_deletion_log_tenant_deleted_idx - ON public.partner_file_deletion_log (tenant_id, deleted_at desc); - -ALTER TABLE public.partner_files ENABLE ROW LEVEL SECURITY; -ALTER TABLE public.partner_files FORCE ROW LEVEL SECURITY; -ALTER TABLE public.partner_file_derived ENABLE ROW LEVEL SECURITY; -ALTER TABLE public.partner_file_derived FORCE ROW LEVEL SECURITY; -ALTER TABLE public.partner_file_deletion_log ENABLE ROW LEVEL SECURITY; -ALTER TABLE public.partner_file_deletion_log FORCE ROW LEVEL SECURITY; - + ON public.partner_file_deletion_log (tenant_id, deleted_at desc) +""", + """ +ALTER TABLE public.partner_files ENABLE ROW LEVEL SECURITY +""", + """ +ALTER TABLE public.partner_files FORCE ROW LEVEL SECURITY +""", + """ +ALTER TABLE public.partner_file_derived ENABLE ROW LEVEL SECURITY +""", + """ +ALTER TABLE public.partner_file_derived FORCE ROW LEVEL SECURITY +""", + """ +ALTER TABLE public.partner_file_deletion_log ENABLE ROW LEVEL SECURITY +""", + """ +ALTER TABLE public.partner_file_deletion_log FORCE ROW LEVEL SECURITY +""", + """ -- CASE form, not OR: public.get_tenant_context() raises when app.tenant_id is -- unset and Postgres does not order OR operands, so the sysadmin arm is tested -- first (same predicate shape as alembic v096 / v134 / v135 / v137). -DROP POLICY IF EXISTS tenant_isolation_partner_files ON public.partner_files; +DROP POLICY IF EXISTS tenant_isolation_partner_files ON public.partner_files +""", + """ CREATE POLICY tenant_isolation_partner_files ON public.partner_files FOR ALL USING ( @@ -196,9 +221,12 @@ AND current_user = 'regengine_sysadmin' THEN true ELSE tenant_id = (select public.get_tenant_context()) END - ); - -DROP POLICY IF EXISTS tenant_isolation_partner_file_derived ON public.partner_file_derived; + ) +""", + """ +DROP POLICY IF EXISTS tenant_isolation_partner_file_derived ON public.partner_file_derived +""", + """ CREATE POLICY tenant_isolation_partner_file_derived ON public.partner_file_derived FOR ALL USING ( @@ -214,9 +242,12 @@ AND current_user = 'regengine_sysadmin' THEN true ELSE tenant_id = (select public.get_tenant_context()) END - ); - -DROP POLICY IF EXISTS tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log; + ) +""", + """ +DROP POLICY IF EXISTS tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log +""", + """ CREATE POLICY tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log FOR SELECT USING ( @@ -225,11 +256,14 @@ AND current_user = 'regengine_sysadmin' THEN true ELSE tenant_id = (select public.get_tenant_context()) END - ); - + ) +""", + """ -- Inserts into the deletion log go through the retention worker (regengine / -- SECURITY DEFINER RPC). No authenticated UPDATE/DELETE on the log. -DROP POLICY IF EXISTS partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log; +DROP POLICY IF EXISTS partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log +""", + """ CREATE POLICY partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log FOR INSERT WITH CHECK ( @@ -238,8 +272,9 @@ AND current_user = 'regengine_sysadmin' THEN true ELSE tenant_id = (select public.get_tenant_context()) END - ); - + ) +""", + """ DO $$ BEGIN IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'anon') THEN @@ -270,11 +305,9 @@ -- Prefer SECURITY DEFINER RPCs; keep service_role from mutating the log. REVOKE UPDATE, DELETE ON public.partner_file_deletion_log FROM service_role; END IF; -END $$; - -""" - -_MANIFEST_BLOBS_SQL = """ +END $$ +""", + """ -- Railway cutover helper: allow hash/pointer-only manifest_blobs rows. DO $$ BEGIN @@ -286,57 +319,31 @@ ) THEN ALTER TABLE fsma.manifest_blobs ALTER COLUMN raw_bytes DROP NOT NULL; END IF; -END $$; - -ALTER TABLE fsma.manifest_blobs DROP CONSTRAINT IF EXISTS manifest_blobs_payload_present_check; +END $$ +""", + """ +ALTER TABLE fsma.manifest_blobs DROP CONSTRAINT IF EXISTS manifest_blobs_payload_present_check +""", + """ ALTER TABLE fsma.manifest_blobs ADD CONSTRAINT manifest_blobs_payload_present_check CHECK ( raw_bytes IS NOT NULL OR (storage_url IS NOT NULL AND length(btrim(storage_url)) > 0) - ); -""" + ) +""", +) def upgrade() -> None: - # Split on semicolons outside dollar-quoted blocks so DO $$ ... $$; stays intact. - def _statements(sql: str) -> list[str]: - out: list[str] = [] - buf: list[str] = [] - in_dollar = False - i = 0 - while i < len(sql): - if sql.startswith("$$", i): - in_dollar = not in_dollar - buf.append("$$") - i += 2 - continue - ch = sql[i] - if ch == ";" and not in_dollar: - stmt = "".join(buf).strip() - if stmt and not all( - line.strip().startswith("--") or not line.strip() - for line in stmt.splitlines() - ): - out.append(stmt) - buf = [] - i += 1 - continue - buf.append(ch) - i += 1 - tail = "".join(buf).strip() - if tail: - out.append(tail) - return out - - for statement in _statements(_PARTNER_FILES_SQL) + _statements(_MANIFEST_BLOBS_SQL): - op.execute(sa.text(statement)) + for statement in _UPGRADE_STATEMENTS: + op.execute(statement) def downgrade() -> None: + op.execute( - sa.text( - """ + """ DO $$ BEGIN IF EXISTS ( @@ -352,18 +359,17 @@ def downgrade() -> None: END IF; END $$; """ - ) ) - op.execute(sa.text("DROP POLICY IF EXISTS partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log")) - op.execute(sa.text("DROP POLICY IF EXISTS tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log")) - op.execute(sa.text("DROP POLICY IF EXISTS tenant_isolation_partner_file_derived ON public.partner_file_derived")) - op.execute(sa.text("DROP POLICY IF EXISTS tenant_isolation_partner_files ON public.partner_files")) - op.execute(sa.text("DROP TRIGGER IF EXISTS partner_files_set_delete_after ON public.partner_files")) - op.execute(sa.text("DROP FUNCTION IF EXISTS public.partner_files_set_delete_after()")) - op.execute(sa.text("DROP INDEX IF EXISTS public.partner_file_deletion_log_tenant_deleted_idx")) - op.execute(sa.text("DROP INDEX IF EXISTS public.partner_file_derived_file_idx")) - op.execute(sa.text("DROP INDEX IF EXISTS public.partner_files_tenant_uploaded_idx")) - op.execute(sa.text("DROP INDEX IF EXISTS public.partner_files_due_delete_idx")) - op.execute(sa.text("DROP TABLE IF EXISTS public.partner_file_deletion_log")) - op.execute(sa.text("DROP TABLE IF EXISTS public.partner_file_derived")) - op.execute(sa.text("DROP TABLE IF EXISTS public.partner_files")) + op.execute("DROP POLICY IF EXISTS partner_file_deletion_log_insert_worker ON public.partner_file_deletion_log") + op.execute("DROP POLICY IF EXISTS tenant_isolation_partner_file_deletion_log ON public.partner_file_deletion_log") + op.execute("DROP POLICY IF EXISTS tenant_isolation_partner_file_derived ON public.partner_file_derived") + op.execute("DROP POLICY IF EXISTS tenant_isolation_partner_files ON public.partner_files") + op.execute("DROP TRIGGER IF EXISTS partner_files_set_delete_after ON public.partner_files") + op.execute("DROP FUNCTION IF EXISTS public.partner_files_set_delete_after()") + op.execute("DROP INDEX IF EXISTS public.partner_file_deletion_log_tenant_deleted_idx") + op.execute("DROP INDEX IF EXISTS public.partner_file_derived_file_idx") + op.execute("DROP INDEX IF EXISTS public.partner_files_tenant_uploaded_idx") + op.execute("DROP INDEX IF EXISTS public.partner_files_due_delete_idx") + op.execute("DROP TABLE IF EXISTS public.partner_file_deletion_log") + op.execute("DROP TABLE IF EXISTS public.partner_file_derived") + op.execute("DROP TABLE IF EXISTS public.partner_files") diff --git a/supabase/migrations/20261028100000_partner_files_storage_retention.sql b/supabase/migrations/20261028100000_partner_files_storage_retention.sql index 6db6e2433..426ec2c41 100644 --- a/supabase/migrations/20261028100000_partner_files_storage_retention.sql +++ b/supabase/migrations/20261028100000_partner_files_storage_retention.sql @@ -77,7 +77,7 @@ $$; DROP TRIGGER IF EXISTS partner_files_set_delete_after ON public.partner_files; CREATE TRIGGER partner_files_set_delete_after - BEFORE INSERT OR UPDATE OF report_delivered_at + BEFORE INSERT OR UPDATE ON public.partner_files FOR EACH ROW EXECUTE FUNCTION public.partner_files_set_delete_after(); diff --git a/tests/migrations/test_v138_partner_file_storage_retention.py b/tests/migrations/test_v138_partner_file_storage_retention.py new file mode 100644 index 000000000..16495bdde --- /dev/null +++ b/tests/migrations/test_v138_partner_file_storage_retention.py @@ -0,0 +1,269 @@ +"""v138 partner file storage retention — structure + stub PG16 behavior. + +Covers the Codex FAIL fixes on #3729: + +1. No runtime semicolon splitter — statements are an explicit tuple (v137 + shape). Structure tests assert a commented ``;`` and a ``$$`` function + body both live inside single statement strings. +2. ``delete_after`` trigger is ``BEFORE INSERT OR UPDATE`` (no ``OF`` list) + and always recomputes from ``report_delivered_at``, so a direct UPDATE of + ``delete_after`` is overwritten. +3. Stub PG16: apply ``upgrade()`` twice (idempotent) against a minimal + schema, then exercise the retention clock. + +Uses a disposable local Postgres database (Homebrew / CI service) via +``scratch_database_engine``. Skips when no Postgres is available; fails under +``REGENGINE_REQUIRE_DB=1``. +""" + +from __future__ import annotations + +import ast +import importlib.util +import os +from pathlib import Path +from uuid import uuid4 + +import pytest +from sqlalchemy import create_engine, text + +from tests.migrations.scratch_db import scratch_database_engine + +REPO_ROOT = Path(__file__).resolve().parents[2] +MIGRATION_PATH = ( + REPO_ROOT + / "alembic" + / "versions" + / "20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py" +) + + +def _load_migration(): + spec = importlib.util.spec_from_file_location( + f"v138_partner_{uuid4().hex}", MIGRATION_PATH + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _statement_strings() -> list[str]: + tree = ast.parse(MIGRATION_PATH.read_text()) + for node in tree.body: + if ( + isinstance(node, ast.AnnAssign) + and isinstance(node.target, ast.Name) + and node.target.id == "_UPGRADE_STATEMENTS" + ): + assert isinstance(node.value, ast.Tuple) + out: list[str] = [] + for elt in node.value.elts: + assert isinstance(elt, ast.Constant) and isinstance(elt.value, str) + out.append(elt.value) + return out + raise AssertionError("_UPGRADE_STATEMENTS not found") + + +def test_no_runtime_semicolon_splitter() -> None: + src = MIGRATION_PATH.read_text() + assert "def _statements" not in src + assert "_UPGRADE_STATEMENTS" in src + tree = ast.parse(src) + names = { + node.name + for node in tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + assert "upgrade" in names and "downgrade" in names + assert "_statements" not in names + + +def test_commented_semicolon_and_dollar_body_are_single_statements() -> None: + """Authoring regression: comment `;` and `$$` bodies must not be split.""" + stmts = _statement_strings() + # Header comment with an intentional semicolon survives inside a statement. + assert any("trigger-maintained; UTC" in s for s in stmts) + # Function body is one statement containing the dollar-quoted block. + fn = [ + s + for s in stmts + if "CREATE OR REPLACE FUNCTION public.partner_files_set_delete_after" in s + ] + assert len(fn) == 1 + assert "AS $$" in fn[0] + assert "interval '30 days'" in fn[0] + assert fn[0].count("$$") >= 2 + # DO $$ role grants stay intact as one statement. + dos = [s for s in stmts if s.lstrip().startswith("DO $$")] + assert dos + assert all("END $$" in s for s in dos) + + +def test_delete_after_trigger_is_before_any_update() -> None: + stmts = _statement_strings() + trig = next(s for s in stmts if "CREATE TRIGGER partner_files_set_delete_after" in s) + assert "BEFORE INSERT OR UPDATE" in trig + assert "OF report_delivered_at" not in trig + + +def _bootstrap_minimal_schema(engine) -> None: + with engine.begin() as conn: + conn.execute( + text( + """ + CREATE TABLE public.tenants ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid() + ); + CREATE OR REPLACE FUNCTION public.get_tenant_context() + RETURNS uuid + LANGUAGE sql + STABLE + AS $$ + SELECT NULLIF(current_setting('app.tenant_id', true), '')::uuid + $$; + CREATE SCHEMA IF NOT EXISTS fsma; + CREATE TABLE fsma.manifest_blobs ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + tenant_id uuid NOT NULL, + raw_bytes bytea NOT NULL, + storage_url text + ); + """ + ) + ) + + +def _apply_upgrade(engine, monkeypatch) -> None: + module = _load_migration() + + class _Operations: + @staticmethod + def execute(statement: str) -> None: + with engine.begin() as conn: + conn.execute(text(statement)) + + monkeypatch.setattr(module, "op", _Operations()) + module.upgrade() + + +@pytest.fixture(scope="module") +def pg_engine(): + """Disposable PG16 database with v138 minimal deps, or skip/fail.""" + # Prefer an explicit scratch URL; otherwise peer-auth local Homebrew PG. + if not os.environ.get("REGENGINE_MIGRATION_TEST_DB"): + os.environ["REGENGINE_MIGRATION_TEST_DB"] = "postgresql://sellers@/postgres?host=/tmp" + db_name = "inc2_scratch_v138_" + uuid4().hex[:12] + with scratch_database_engine(db_name) as engine: + _bootstrap_minimal_schema(engine) + yield engine + + +def test_upgrade_idempotent_on_stub_pg16(pg_engine, monkeypatch) -> None: + _apply_upgrade(pg_engine, monkeypatch) + _apply_upgrade(pg_engine, monkeypatch) # second apply must be a no-op + with pg_engine.connect() as conn: + exists = conn.execute( + text( + """ + SELECT COUNT(*) FROM information_schema.tables + WHERE table_schema = 'public' + AND table_name IN ( + 'partner_files', + 'partner_file_derived', + 'partner_file_deletion_log' + ) + """ + ) + ).scalar_one() + assert exists == 3 + version = conn.execute(text("SHOW server_version")).scalar_one() + assert version.startswith("16."), version + tgtype = conn.execute( + text( + """ + SELECT pg_get_triggerdef(t.oid) + FROM pg_trigger t + JOIN pg_class c ON c.oid = t.tgrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' + AND c.relname = 'partner_files' + AND t.tgname = 'partner_files_set_delete_after' + AND NOT t.tgisinternal + """ + ) + ).scalar_one() + assert "BEFORE INSERT OR UPDATE" in tgtype + assert "OF report_delivered_at" not in tgtype + + +def test_delete_after_recomputed_on_direct_update(pg_engine, monkeypatch) -> None: + # Ensure schema is present (idempotent with the previous test). + _apply_upgrade(pg_engine, monkeypatch) + tenant_id = str(uuid4()) + file_id = str(uuid4()) + engagement_id = str(uuid4()) + with pg_engine.begin() as conn: + conn.execute( + text("INSERT INTO public.tenants (id) VALUES (CAST(:id AS uuid))"), + {"id": tenant_id}, + ) + # Bypass RLS for the owning role on a fresh stub. + conn.execute(text("ALTER TABLE public.partner_files DISABLE ROW LEVEL SECURITY")) + row = conn.execute( + text( + """ + INSERT INTO public.partner_files ( + id, tenant_id, engagement_id, storage_path, content_sha256, + original_filename, byte_size, content_type, status, + report_delivered_at + ) VALUES ( + CAST(:id AS uuid), + CAST(:tenant_id AS uuid), + CAST(:engagement_id AS uuid), + 'tenants/' || :tenant_id || '/synthetic.csv', + repeat('a', 64), + 'synthetic.csv', + 12, + 'text/csv', + 'delivered', + TIMESTAMPTZ '2026-10-01 00:00:00+00' + ) + RETURNING delete_after + """ + ), + { + "id": file_id, + "tenant_id": tenant_id, + "engagement_id": engagement_id, + }, + ).scalar_one() + assert row.utctimetuple()[:3] == (2026, 10, 31) + + # Direct UPDATE of delete_after must be overwritten by the trigger. + overwritten = conn.execute( + text( + """ + UPDATE public.partner_files + SET delete_after = TIMESTAMPTZ '2099-01-01 00:00:00+00' + WHERE id = CAST(:id AS uuid) + RETURNING delete_after AT TIME ZONE 'UTC' + """ + ), + {"id": file_id}, + ).scalar_one() + assert str(overwritten)[:10] == "2026-10-31" + + # Clearing delivery clears the clock. + cleared = conn.execute( + text( + """ + UPDATE public.partner_files + SET report_delivered_at = NULL + WHERE id = CAST(:id AS uuid) + RETURNING delete_after + """ + ), + {"id": file_id}, + ).scalar_one() + assert cleared is None From 06212b651d62928169d827e6d78f2a43af214753 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 28 Oct 2026 08:53:29 -0700 Subject: [PATCH 3/6] feat(storage): cut over manifest uploads to partner-uploads pointers When object storage is configured, remediation and webhook_compat write bytes to the private partner-uploads bucket under tenant-scoped paths, record public.partner_files metadata, leave fsma.manifest_blobs.raw_bytes NULL with storage_url set, and emit a partner_file.upload audit event. get_run_meta resolves pointer-only rows from storage. Synthetic tests only. Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- .../ingestion/app/partner_file_storage.py | 208 ++++++++++++++++++ services/ingestion/app/remediation/service.py | 76 ++----- services/ingestion/app/webhook_compat.py | 33 +-- .../test_partner_file_storage_cutover.py | 156 +++++++++++++ 4 files changed, 399 insertions(+), 74 deletions(-) create mode 100644 services/ingestion/app/partner_file_storage.py create mode 100644 services/ingestion/tests/test_partner_file_storage_cutover.py diff --git a/services/ingestion/app/partner_file_storage.py b/services/ingestion/app/partner_file_storage.py new file mode 100644 index 000000000..37e95041e --- /dev/null +++ b/services/ingestion/app/partner_file_storage.py @@ -0,0 +1,208 @@ +"""First-file partner upload persistence (SEL-549 / partner-uploads). + +Uploads land in the private ``partner-uploads`` bucket under a tenant-scoped +path. Metadata is recorded in ``public.partner_files``; ``fsma.manifest_blobs`` +keeps the SHA-256 + ``storage_url`` pointer and leaves ``raw_bytes`` NULL when +object storage is configured. + +Synthetic / test fixtures only until SEL-590 clears. Do not invent partner +payloads; callers pass bytes they already hold. +""" + +from __future__ import annotations + +import hashlib +import logging +import re +from typing import Any, Dict, Optional +from uuid import uuid4 + +from fastapi import HTTPException +from sqlalchemy import text + +logger = logging.getLogger("partner_file_storage") + +PARTNER_UPLOADS_BUCKET = "partner-uploads" +_SAFE_NAME = re.compile(r"[^A-Za-z0-9._-]+") + + +def tenant_object_key( + *, + tenant_id: str, + engagement_id: str, + filename: str, + object_id: Optional[str] = None, +) -> str: + """Build ``tenants/{tenant}/engagements/{engagement}/{uuid}-{filename}``.""" + from .s3_utils import safe_s3_key_component, _validate_s3_key + + oid = object_id or str(uuid4()) + safe_name = safe_s3_key_component(filename, fallback="upload.csv") + key = ( + f"tenants/{safe_s3_key_component(tenant_id, fallback='tenant')}" + f"/engagements/{safe_s3_key_component(engagement_id, fallback='engagement')}" + f"/{safe_s3_key_component(oid, fallback='object')}-{safe_name}" + ) + _validate_s3_key(key) + return key + + +def persist_partner_upload( + db, + *, + tenant_id: str, + engagement_id: str, + content: bytes, + filename: str, + content_type: Optional[str], + uploaded_by: Optional[str], + ingestion_run_id: str, +) -> Dict[str, Any]: + """Store bytes in partner-uploads (when configured) and write metadata + blob pointer. + + Returns a dict with ``partner_file_id``, ``storage_url``, ``storage_path``, + ``sha256``, and ``used_object_storage``. + """ + from .remediation.audit_utils import audit_event + from .s3_utils import is_object_storage_configured, put_bytes + + if not content: + raise HTTPException(status_code=400, detail="Empty upload") + + sha256 = hashlib.sha256(content).hexdigest() + ctype = content_type or "text/csv" + partner_file_id = str(uuid4()) + storage_path = tenant_object_key( + tenant_id=tenant_id, + engagement_id=engagement_id, + filename=filename, + object_id=partner_file_id, + ) + used_object_storage = False + storage_url: Optional[str] = None + raw_bytes_param: Optional[bytes] = content + + if is_object_storage_configured(): + storage_url = put_bytes( + PARTNER_UPLOADS_BUCKET, + storage_path, + content, + ctype, + ) + used_object_storage = True + raw_bytes_param = None # pointer-only row; bytes live in the bucket + db.execute( + # nosemgrep: avoid-sqlalchemy-text — parameterized with :param + text( + """ + INSERT INTO public.partner_files ( + id, tenant_id, engagement_id, storage_bucket, storage_path, + content_sha256, byte_size, content_type, original_filename, + status + ) VALUES ( + CAST(:id AS uuid), + CAST(:tenant_id AS uuid), + CAST(:engagement_id AS uuid), + :bucket, + :storage_path, + :sha256, + :byte_size, + :content_type, + :filename, + 'uploaded' + ) + """ + ), + { + "id": partner_file_id, + "tenant_id": tenant_id, + "engagement_id": engagement_id, + "bucket": PARTNER_UPLOADS_BUCKET, + "storage_path": storage_path, + "sha256": sha256, + "byte_size": len(content), + "content_type": ctype, + "filename": filename, + }, + ) + + db.execute( + # nosemgrep: avoid-sqlalchemy-text — parameterized with :param + text( + """ + INSERT INTO fsma.manifest_blobs + (ingestion_run_id, tenant_id, raw_bytes, sha256, + original_filename, content_type, uploaded_by, storage_url) + VALUES + (CAST(:run_id AS uuid), CAST(:tenant_id AS uuid), :raw_bytes, + :sha256, :filename, :content_type, :uploaded_by, :storage_url) + """ + ), + { + "run_id": ingestion_run_id, + "tenant_id": tenant_id, + "raw_bytes": raw_bytes_param, + "sha256": sha256, + "filename": filename, + "content_type": ctype, + "uploaded_by": uploaded_by, + "storage_url": storage_url, + }, + ) + + audit_event( + db, + tenant_id=tenant_id, + event_type="partner_file.upload", + action="partner_file.upload", + severity="info", + actor_id=uploaded_by, + resource_type="partner_file" if used_object_storage else "manifest_blob", + resource_id=partner_file_id if used_object_storage else ingestion_run_id, + metadata={ + "ingestion_run_id": ingestion_run_id, + "engagement_id": engagement_id, + "storage_bucket": PARTNER_UPLOADS_BUCKET if used_object_storage else None, + "storage_path": storage_path if used_object_storage else None, + "storage_url": storage_url, + "content_sha256": sha256, + "byte_size": len(content), + "original_filename": filename, + "used_object_storage": used_object_storage, + }, + ) + + return { + "partner_file_id": partner_file_id if used_object_storage else None, + "storage_url": storage_url, + "storage_path": storage_path if used_object_storage else None, + "sha256": sha256, + "used_object_storage": used_object_storage, + } + + +def load_bytes_from_storage_url(storage_url: str) -> bytes: + """Fetch object bytes for an ``s3://bucket/key`` storage_url.""" + from .s3_utils import get_raw_document + + if not storage_url.startswith("s3://"): + raise HTTPException(status_code=500, detail="Unsupported storage_url scheme") + without = storage_url[len("s3://") :] + bucket, _, key = without.partition("/") + if not bucket or not key: + raise HTTPException(status_code=500, detail="Malformed storage_url") + doc = get_raw_document(key=key, bucket=bucket) + return doc["content"] + + +def resolve_manifest_raw_bytes( + *, + raw_bytes: Any, + storage_url: Optional[str], +) -> bytes: + """Return manifest bytes from bytea or, when null, from storage_url.""" + if raw_bytes is not None: + return bytes(raw_bytes) + if storage_url: + return load_bytes_from_storage_url(storage_url) + raise HTTPException(status_code=404, detail="Manifest payload not found") diff --git a/services/ingestion/app/remediation/service.py b/services/ingestion/app/remediation/service.py index 06d8db830..1c35d7bc7 100644 --- a/services/ingestion/app/remediation/service.py +++ b/services/ingestion/app/remediation/service.py @@ -109,53 +109,21 @@ def ingest_manifest( ) run_id = store.create_ingestion_run(run) - db.execute( - # nosemgrep: avoid-sqlalchemy-text — parameterized with :param - text( - """ - INSERT INTO fsma.manifest_blobs - (ingestion_run_id, tenant_id, raw_bytes, sha256, - original_filename, content_type, uploaded_by) - VALUES - (CAST(:run_id AS uuid), CAST(:tenant_id AS uuid), :raw_bytes, - :sha256, :filename, :content_type, :uploaded_by) - """ - ), - { - "run_id": run_id, - "tenant_id": tenant_id, - "raw_bytes": raw_bytes, - "sha256": sha256, - "filename": filename, - "content_type": content_type, - "uploaded_by": uploaded_by, - }, - ) + # First-file cutover: prefer private partner-uploads (pointer-only + # manifest_blobs + partner_files metadata + audit). Falls back to + # bytea when object storage is not configured (local/dev). + from ..partner_file_storage import persist_partner_upload - # Optional off-box mirror; the bytea row remains the authoritative copy. - try: - from ..s3_utils import RAW_INGEST_BUCKET, is_object_storage_configured, put_bytes - - if is_object_storage_configured(): - storage_url = put_bytes( - RAW_INGEST_BUCKET, - f"remediation/{tenant_id}/{run_id}/{filename}", - raw_bytes, - content_type or "text/csv", - ) - db.execute( - # nosemgrep: avoid-sqlalchemy-text — parameterized with :param - text( - """ - UPDATE fsma.manifest_blobs SET storage_url = :storage_url - WHERE ingestion_run_id = CAST(:run_id AS uuid) - AND tenant_id = CAST(:tenant_id AS uuid) - """ - ), - {"storage_url": storage_url, "run_id": run_id, "tenant_id": tenant_id}, - ) - except Exception as exc: # pragma: no cover — mirror is best-effort - logger.warning("manifest_blob_mirror_failed run=%s error=%s", run_id, exc) + persist_partner_upload( + db, + tenant_id=tenant_id, + engagement_id=run_id, + content=raw_bytes, + filename=filename, + content_type=content_type, + uploaded_by=uploaded_by, + ingestion_run_id=run_id, + ) seen_fingerprints: Dict[str, int] = {} records: List[Dict[str, Any]] = [] @@ -787,12 +755,14 @@ def release_record(db, *, tenant_id: str, record: Dict[str, Any], source_meta: D def get_run_meta(db, *, tenant_id: str, run_id: str) -> Dict[str, Any]: + from ..partner_file_storage import resolve_manifest_raw_bytes + row = db.execute( # nosemgrep: avoid-sqlalchemy-text — parameterized with :param text( """ - SELECT b.sha256, b.original_filename, b.raw_bytes, r.status, - r.record_count, r.accepted_count, r.rejected_count + SELECT b.sha256, b.original_filename, b.raw_bytes, b.storage_url, + r.status, r.record_count, r.accepted_count, r.rejected_count FROM fsma.manifest_blobs b JOIN fsma.ingestion_runs r ON r.id = b.ingestion_run_id WHERE b.ingestion_run_id = CAST(:run_id AS uuid) @@ -807,11 +777,11 @@ def get_run_meta(db, *, tenant_id: str, run_id: str) -> Dict[str, Any]: return { "sha256": row[0], "filename": row[1], - "raw_bytes": bytes(row[2]), - "run_status": row[3], - "record_count": row[4], - "accepted_count": row[5], - "rejected_count": row[6], + "raw_bytes": resolve_manifest_raw_bytes(raw_bytes=row[2], storage_url=row[3]), + "run_status": row[4], + "record_count": row[5], + "accepted_count": row[6], + "rejected_count": row[7], } diff --git a/services/ingestion/app/webhook_compat.py b/services/ingestion/app/webhook_compat.py index 71cb3964b..17a7da327 100644 --- a/services/ingestion/app/webhook_compat.py +++ b/services/ingestion/app/webhook_compat.py @@ -173,27 +173,18 @@ def _open_ingestion_run(db_session, tenant_id: str, source: str, receipt: Source initiated_by=receipt.initiated_by, ) run_id = store.create_ingestion_run(run) - db_session.execute( - # nosemgrep: avoid-sqlalchemy-text — parameterized with :param - text( - """ - INSERT INTO fsma.manifest_blobs - (ingestion_run_id, tenant_id, raw_bytes, sha256, - original_filename, content_type, uploaded_by) - VALUES - (CAST(:run_id AS uuid), CAST(:tenant_id AS uuid), :raw_bytes, - :sha256, :filename, :content_type, :uploaded_by) - """ - ), - { - "run_id": run_id, - "tenant_id": tenant_id, - "raw_bytes": receipt.raw_bytes, - "sha256": receipt.sha256, - "filename": receipt.filename, - "content_type": receipt.content_type, - "uploaded_by": receipt.initiated_by, - }, + # First-file cutover: partner-uploads + pointer when object storage is on. + from .partner_file_storage import persist_partner_upload + + persist_partner_upload( + db_session, + tenant_id=tenant_id, + engagement_id=run_id, + content=receipt.raw_bytes, + filename=receipt.filename, + content_type=receipt.content_type, + uploaded_by=receipt.initiated_by, + ingestion_run_id=run_id, ) return run_id diff --git a/services/ingestion/tests/test_partner_file_storage_cutover.py b/services/ingestion/tests/test_partner_file_storage_cutover.py new file mode 100644 index 000000000..e5f3eada0 --- /dev/null +++ b/services/ingestion/tests/test_partner_file_storage_cutover.py @@ -0,0 +1,156 @@ +"""Unit tests for first-file partner-uploads cutover (synthetic data only).""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional +from uuid import uuid4 + +import pytest + +from services.ingestion.app import partner_file_storage as pfs + + +class _FakeResult: + def __init__(self, row=None): + self._row = row + + def fetchone(self): + return self._row + + def scalar_one(self): + return self._row + + +class _FakeDB: + def __init__(self): + self.statements: List[str] = [] + self.params: List[Dict[str, Any]] = [] + + def execute(self, statement, params=None): + sql = str(statement) + self.statements.append(sql) + self.params.append(params or {}) + return _FakeResult() + + def begin_nested(self): + from contextlib import nullcontext + + return nullcontext() + + +@pytest.fixture +def tenant_id() -> str: + return str(uuid4()) + + +def test_tenant_object_key_is_tenant_scoped(tenant_id: str) -> None: + engagement = str(uuid4()) + key = pfs.tenant_object_key( + tenant_id=tenant_id, + engagement_id=engagement, + filename="synth-partner.csv", + object_id="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", + ) + assert key.startswith(f"tenants/{tenant_id}/engagements/{engagement}/") + assert key.endswith("-synth-partner.csv") + assert ".." not in key + + +def test_persist_uses_partner_uploads_and_null_raw_bytes(monkeypatch, tenant_id: str) -> None: + put_calls: List[Dict[str, Any]] = [] + audits: List[Dict[str, Any]] = [] + + monkeypatch.setattr(pfs, "PARTNER_UPLOADS_BUCKET", "partner-uploads") + + def _is_configured() -> bool: + return True + + def _put_bytes(bucket, key, content, content_type="application/octet-stream"): + put_calls.append( + {"bucket": bucket, "key": key, "size": len(content), "content_type": content_type} + ) + return f"s3://{bucket}/{key}" + + # Patch through the import sites used inside persist_partner_upload + import services.ingestion.app.s3_utils as s3_utils + + monkeypatch.setattr(s3_utils, "is_object_storage_configured", _is_configured) + monkeypatch.setattr(s3_utils, "put_bytes", _put_bytes) + + import services.ingestion.app.remediation.audit_utils as audit_utils + + def _audit(db, **kwargs): + audits.append(kwargs) + + monkeypatch.setattr(audit_utils, "audit_event", _audit) + + db = _FakeDB() + run_id = str(uuid4()) + content = b"lot,qty\nsynth-1,1\n" # synthetic only + result = pfs.persist_partner_upload( + db, + tenant_id=tenant_id, + engagement_id=run_id, + content=content, + filename="synth-partner.csv", + content_type="text/csv", + uploaded_by=None, + ingestion_run_id=run_id, + ) + + assert result["used_object_storage"] is True + assert result["storage_url"].startswith("s3://partner-uploads/tenants/") + assert put_calls and put_calls[0]["bucket"] == "partner-uploads" + # partner_files INSERT then manifest_blobs INSERT + assert any("INSERT INTO public.partner_files" in s for s in db.statements) + blob_params = next(p for p, s in zip(db.params, db.statements) if "manifest_blobs" in s) + assert blob_params["raw_bytes"] is None + assert blob_params["storage_url"].startswith("s3://partner-uploads/") + assert audits and audits[0]["event_type"] == "partner_file.upload" + assert audits[0]["metadata"]["used_object_storage"] is True + + +def test_persist_falls_back_to_bytea_without_object_storage(monkeypatch, tenant_id: str) -> None: + import services.ingestion.app.s3_utils as s3_utils + import services.ingestion.app.remediation.audit_utils as audit_utils + + monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: False) + monkeypatch.setattr(audit_utils, "audit_event", lambda *a, **k: None) + + db = _FakeDB() + run_id = str(uuid4()) + content = b"a,b\n1,2\n" + result = pfs.persist_partner_upload( + db, + tenant_id=tenant_id, + engagement_id=run_id, + content=content, + filename="local.csv", + content_type="text/csv", + uploaded_by="system", + ingestion_run_id=run_id, + ) + assert result["used_object_storage"] is False + assert result["storage_url"] is None + assert not any("partner_files" in s for s in db.statements) + blob_params = next(p for p, s in zip(db.params, db.statements) if "manifest_blobs" in s) + assert blob_params["raw_bytes"] == content + + +def test_resolve_manifest_raw_bytes_prefers_bytea() -> None: + assert pfs.resolve_manifest_raw_bytes(raw_bytes=b"abc", storage_url=None) == b"abc" + + +def test_resolve_manifest_raw_bytes_fetches_storage(monkeypatch) -> None: + import services.ingestion.app.s3_utils as s3_utils + + monkeypatch.setattr( + s3_utils, + "get_raw_document", + lambda key, bucket: {"content": b"from-s3"}, + ) + out = pfs.resolve_manifest_raw_bytes( + raw_bytes=None, + storage_url="s3://partner-uploads/tenants/t/engagements/e/obj-file.csv", + ) + assert out == b"from-s3" From 514ad7c75ca7acdbfba7c17909afd53f7df8abb9 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 28 Oct 2026 09:04:51 -0700 Subject: [PATCH 4/6] fix(storage): grant runtime INSERT on manifest_blobs.storage_url Claude FAIL on #3732: partner-uploads cutover writes storage_url, but v126 column INSERT excluded it. Extend v138 + v111 inventory so the runtime role can insert pointer-only rows; keep created_at ungated. Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- ...3_b71d9e4c2a60_v111_runtime_role_grants.py | 9 ++--- ...901_v138_partner_file_storage_retention.py | 34 +++++++++++++++++++ ...100000_partner_files_storage_retention.sql | 17 ++++++++++ ...est_v138_partner_file_storage_retention.py | 23 +++++++++++++ .../test_remediation_disabled_contract_pg.py | 8 ++--- .../test_postgres_evidence_workflow_pg.py | 18 +++++----- 6 files changed, 92 insertions(+), 17 deletions(-) diff --git a/alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py b/alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py index 48d3f91c5..960778696 100644 --- a/alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py +++ b/alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py @@ -652,10 +652,10 @@ def _build_migration_owner_posture() -> str: # plane data. "public.tenants": {"SELECT": ("id",)}, # v126 -- POST /api/v1/ingest/csv (#3122) retains each upload verbatim - # in the ingest transaction (webhook_compat._open_ingestion_run). INSERT - # is limited to the columns that statement writes, so storage_url and - # created_at cannot be forged; reads, updates and deletes stay denied - # (the disabled remediation surface keeps its contract). The composite + # in the ingest transaction (webhook_compat._open_ingestion_run). + # v138 extends INSERT with storage_url for partner-uploads pointer rows; + # created_at stays ungated so timestamps cannot be forged. Reads, updates + # and deletes stay denied (disabled remediation surface). The composite # (ingestion_run_id, tenant_id) foreign key pins a blob to its own # tenant's run. "fsma.manifest_blobs": { @@ -667,6 +667,7 @@ def _build_migration_owner_posture() -> str: "original_filename", "content_type", "uploaded_by", + "storage_url", ) }, # v126 -- the agent-artifact worker (shared.agent_task_bridge) reads only diff --git a/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py b/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py index 01b09b86d..532bed829 100644 --- a/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py +++ b/alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py @@ -331,6 +331,23 @@ raw_bytes IS NOT NULL OR (storage_url IS NOT NULL AND length(btrim(storage_url)) > 0) ) +""", + """ +DO $$ +BEGIN + -- v138: runtime role must INSERT storage_url for partner-uploads pointer rows. + -- Inventory (v111 REGENGINE_COLUMN_GRANTS) also lists storage_url so + -- apply_runtime_role_cutover.py keeps the grant across deploys. + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine') + AND EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'fsma' + AND table_name = 'manifest_blobs' + AND column_name = 'storage_url' + ) THEN + GRANT INSERT (storage_url) ON TABLE fsma.manifest_blobs TO regengine; + END IF; +END $$ """, ) @@ -345,6 +362,23 @@ def downgrade() -> None: op.execute( """ DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine') + AND EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'fsma' + AND table_name = 'manifest_blobs' + AND column_name = 'storage_url' + ) THEN + REVOKE INSERT (storage_url) ON TABLE fsma.manifest_blobs FROM regengine; + END IF; +END $$; +""" + ) + + op.execute( + """ +DO $$ BEGIN IF EXISTS ( SELECT 1 FROM information_schema.tables diff --git a/supabase/migrations/20261028100000_partner_files_storage_retention.sql b/supabase/migrations/20261028100000_partner_files_storage_retention.sql index 426ec2c41..1b006d6b1 100644 --- a/supabase/migrations/20261028100000_partner_files_storage_retention.sql +++ b/supabase/migrations/20261028100000_partner_files_storage_retention.sql @@ -254,3 +254,20 @@ END $$; -- DROP TABLE IF EXISTS public.partner_file_deletion_log; -- DROP TABLE IF EXISTS public.partner_file_derived; -- DROP TABLE IF EXISTS public.partner_files; + +-- Runtime role: allow pointer-only manifest_blobs inserts (partner-uploads). +DO $$ +BEGIN + -- v138: runtime role must INSERT storage_url for partner-uploads pointer rows. + -- Inventory (v111 REGENGINE_COLUMN_GRANTS) also lists storage_url so + -- apply_runtime_role_cutover.py keeps the grant across deploys. + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'regengine') + AND EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'fsma' + AND table_name = 'manifest_blobs' + AND column_name = 'storage_url' + ) THEN + GRANT INSERT (storage_url) ON TABLE fsma.manifest_blobs TO regengine; + END IF; +END $$; diff --git a/tests/migrations/test_v138_partner_file_storage_retention.py b/tests/migrations/test_v138_partner_file_storage_retention.py index 16495bdde..2a5bf12e3 100644 --- a/tests/migrations/test_v138_partner_file_storage_retention.py +++ b/tests/migrations/test_v138_partner_file_storage_retention.py @@ -267,3 +267,26 @@ def test_delete_after_recomputed_on_direct_update(pg_engine, monkeypatch) -> Non {"id": file_id}, ).scalar_one() assert cleared is None + +def test_runtime_grant_includes_storage_url_insert() -> None: + """v138 must GRANT INSERT (storage_url) and inventory must list it. + + Structural only — no live DB / no partner payloads. + """ + from pathlib import Path + import importlib.util + + alembic = Path("alembic/versions/20261028_b8c1e4f2a901_v138_partner_file_storage_retention.py") + text = alembic.read_text(encoding="utf-8") + assert "GRANT INSERT (storage_url) ON TABLE fsma.manifest_blobs TO regengine" in text + assert "REVOKE INSERT (storage_url) ON TABLE fsma.manifest_blobs FROM regengine" in text + + inv_path = Path("alembic/versions/20260803_b71d9e4c2a60_v111_runtime_role_grants.py") + spec = importlib.util.spec_from_file_location("v111_grants", inv_path) + mod = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(mod) + cols = mod.REGENGINE_COLUMN_GRANTS["fsma.manifest_blobs"]["INSERT"] + assert "storage_url" in cols + assert "created_at" not in cols + diff --git a/tests/security/test_remediation_disabled_contract_pg.py b/tests/security/test_remediation_disabled_contract_pg.py index 06e18311e..4ad14760d 100644 --- a/tests/security/test_remediation_disabled_contract_pg.py +++ b/tests/security/test_remediation_disabled_contract_pg.py @@ -17,10 +17,10 @@ from tests.migrations.rls_db import guarded_live_database_url PROHIBITED_TABLE_PRIVILEGES = { - # The enabled CSV upload route retains each file here through a v126 - # column-scoped INSERT that excludes storage_url and created_at. No - # table-wide privilege exists: reading, re-pointing or deleting a - # retained manifest is remediation work and stays denied. + # The enabled CSV upload route retains each file here through a + # column-scoped INSERT (v126 columns + v138 storage_url). created_at stays + # excluded. No table-wide privilege exists: reading, re-pointing (UPDATE) + # or deleting a retained manifest is remediation work and stays denied. "fsma.manifest_blobs": ("SELECT", "INSERT", "UPDATE", "DELETE"), "fsma.manifest_records": ("SELECT", "INSERT", "UPDATE", "DELETE"), "fsma.remediation_cases": ("SELECT", "INSERT", "UPDATE", "DELETE"), diff --git a/tests/shared/test_postgres_evidence_workflow_pg.py b/tests/shared/test_postgres_evidence_workflow_pg.py index c851a77bb..963720050 100644 --- a/tests/shared/test_postgres_evidence_workflow_pg.py +++ b/tests/shared/test_postgres_evidence_workflow_pg.py @@ -556,15 +556,15 @@ def rejected(match, statement, params=None, call=None): blob_insert, {"run": other_blobless_run, "tenant": w.tenant, "sha": "0" * 64}, ) - # INSERT is column-scoped: retention metadata cannot be forged. - for forged_column, forged_value in (("storage_url", "https://attacker.example/x"), ("created_at", "1990-01-01")): - rejected( - "permission denied", - f"""INSERT INTO fsma.manifest_blobs - (ingestion_run_id, tenant_id, raw_bytes, sha256, original_filename, {forged_column}) - VALUES (CAST(:run AS uuid), CAST(:tenant AS uuid), '\\x00', :sha, 'forged.csv', :value)""", - {"run": own_run, "tenant": w.tenant, "sha": "0" * 64, "value": forged_value}, - ) + # INSERT is column-scoped: created_at still cannot be forged. v138 grants + # storage_url INSERT for partner-uploads pointer-only rows. + rejected( + "permission denied", + """INSERT INTO fsma.manifest_blobs + (ingestion_run_id, tenant_id, raw_bytes, sha256, original_filename, created_at) + VALUES (CAST(:run AS uuid), CAST(:tenant AS uuid), '\\x00', :sha, 'forged.csv', :value)""", + {"run": own_run, "tenant": w.tenant, "sha": "0" * 64, "value": "1990-01-01"}, + ) rejected("row-level security", None, call=lambda db: persist_agent_artifact(db, other_payload)) # Reusing another tenant's artifact id under the own tenant is a loud # primary-key violation, never a silent ON CONFLICT no-op. From 6859ff32a3b202c3d65d672d4d0e2c9541aef46e Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 28 Oct 2026 09:05:57 -0700 Subject: [PATCH 5/6] fix(storage): fail closed when partner object storage is off Claude FAIL #3732: missing storage credentials no longer silently write partner CSV bytes into manifest_blobs.raw_bytes. Production always refuses (HTTP 503). Local/test may opt in with REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK=1. Schema grant for storage_url comes from merged #3729. Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- .../ingestion/app/partner_file_storage.py | 42 +++++++++++++- .../test_partner_file_storage_cutover.py | 55 ++++++++++++++++++- 2 files changed, 94 insertions(+), 3 deletions(-) diff --git a/services/ingestion/app/partner_file_storage.py b/services/ingestion/app/partner_file_storage.py index 37e95041e..6f2e1a699 100644 --- a/services/ingestion/app/partner_file_storage.py +++ b/services/ingestion/app/partner_file_storage.py @@ -12,6 +12,7 @@ from __future__ import annotations import hashlib +import os import logging import re from typing import Any, Dict, Optional @@ -23,6 +24,22 @@ logger = logging.getLogger("partner_file_storage") PARTNER_UPLOADS_BUCKET = "partner-uploads" + +def allow_manifest_bytea_fallback() -> bool: + """Permit raw_bytes fallback only in non-production with an explicit flag. + + Production always fail-closes when object storage is not configured so + partner CSV payloads cannot silently land in Postgres. Local/test must set + ``REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK=1`` to opt into the legacy path. + """ + from shared.env import is_production + + if is_production(): + return False + flag = os.getenv("REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK", "").strip().lower() + return flag in {"1", "true", "yes", "on"} + + _SAFE_NAME = re.compile(r"[^A-Za-z0-9._-]+") @@ -58,7 +75,11 @@ def persist_partner_upload( uploaded_by: Optional[str], ingestion_run_id: str, ) -> Dict[str, Any]: - """Store bytes in partner-uploads (when configured) and write metadata + blob pointer. + """Store bytes in partner-uploads and write metadata + blob pointer. + + Production fail-closes with HTTP 503 when object storage is not configured. + Local/test may set ``REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK=1`` to keep the + legacy ``manifest_blobs.raw_bytes`` path. Returns a dict with ``partner_file_id``, ``storage_url``, ``storage_path``, ``sha256``, and ``used_object_storage``. @@ -80,7 +101,7 @@ def persist_partner_upload( ) used_object_storage = False storage_url: Optional[str] = None - raw_bytes_param: Optional[bytes] = content + raw_bytes_param: Optional[bytes] = None if is_object_storage_configured(): storage_url = put_bytes( @@ -125,6 +146,23 @@ def persist_partner_upload( "filename": filename, }, ) + else: + if not allow_manifest_bytea_fallback(): + logger.error( + "partner_upload_storage_unconfigured tenant=%s run=%s", + tenant_id, + ingestion_run_id, + ) + raise HTTPException( + status_code=503, + detail=( + "Partner object storage is not configured. " + "Refusing to store upload bytes in the database." + ), + ) + # Explicit local/test fallback only. + raw_bytes_param = content + storage_url = None db.execute( # nosemgrep: avoid-sqlalchemy-text — parameterized with :param diff --git a/services/ingestion/tests/test_partner_file_storage_cutover.py b/services/ingestion/tests/test_partner_file_storage_cutover.py index e5f3eada0..491c3c9ec 100644 --- a/services/ingestion/tests/test_partner_file_storage_cutover.py +++ b/services/ingestion/tests/test_partner_file_storage_cutover.py @@ -110,10 +110,12 @@ def _audit(db, **kwargs): assert audits[0]["metadata"]["used_object_storage"] is True -def test_persist_falls_back_to_bytea_without_object_storage(monkeypatch, tenant_id: str) -> None: +def test_persist_falls_back_to_bytea_with_explicit_local_flag(monkeypatch, tenant_id: str) -> None: import services.ingestion.app.s3_utils as s3_utils import services.ingestion.app.remediation.audit_utils as audit_utils + monkeypatch.setenv("REGENGINE_ENV", "development") + monkeypatch.setenv("REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK", "1") monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: False) monkeypatch.setattr(audit_utils, "audit_event", lambda *a, **k: None) @@ -137,6 +139,57 @@ def test_persist_falls_back_to_bytea_without_object_storage(monkeypatch, tenant_ assert blob_params["raw_bytes"] == content +def test_persist_fail_closed_without_storage_or_flag(monkeypatch, tenant_id: str) -> None: + """Without object storage and without the local flag, refuse the upload.""" + import services.ingestion.app.s3_utils as s3_utils + from fastapi import HTTPException + + monkeypatch.setenv("REGENGINE_ENV", "development") + monkeypatch.delenv("REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK", raising=False) + monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: False) + + db = _FakeDB() + run_id = str(uuid4()) + with pytest.raises(HTTPException) as ei: + pfs.persist_partner_upload( + db, + tenant_id=tenant_id, + engagement_id=run_id, + content=b"a,b\n1,2\n", + filename="local.csv", + content_type="text/csv", + uploaded_by="system", + ingestion_run_id=run_id, + ) + assert ei.value.status_code == 503 + assert not db.statements + + +def test_persist_fail_closed_in_production_even_with_flag(monkeypatch, tenant_id: str) -> None: + """Production never allows the bytea fallback, flag or not.""" + import services.ingestion.app.s3_utils as s3_utils + from fastapi import HTTPException + + monkeypatch.setenv("REGENGINE_ENV", "production") + monkeypatch.setenv("REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK", "1") + monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: False) + + db = _FakeDB() + run_id = str(uuid4()) + with pytest.raises(HTTPException) as ei: + pfs.persist_partner_upload( + db, + tenant_id=tenant_id, + engagement_id=run_id, + content=b"a,b\n1,2\n", + filename="local.csv", + content_type="text/csv", + uploaded_by="system", + ingestion_run_id=run_id, + ) + assert ei.value.status_code == 503 + + def test_resolve_manifest_raw_bytes_prefers_bytea() -> None: assert pfs.resolve_manifest_raw_bytes(raw_bytes=b"abc", storage_url=None) == b"abc" From dcaea7bcec0b49550edbadbb2a102e2c368ad43e Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:06:41 -0700 Subject: [PATCH 6/6] test(ingestion): pin storage mode for partner-upload cutover tests _open_ingestion_run now stores bytes through persist_partner_upload, which fails closed without object storage. The unit test pins the explicit local bytea fallback, and the live-DB CI step (no object storage) opts into the same non-production flag so the CSV runtime-role tests still run. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/backend-ci.yml | 4 ++++ services/ingestion/tests/test_webhook_compat_1342.py | 11 +++++++++++ 2 files changed, 15 insertions(+) diff --git a/.github/workflows/backend-ci.yml b/.github/workflows/backend-ci.yml index 02f181371..12839ea0c 100644 --- a/.github/workflows/backend-ci.yml +++ b/.github/workflows/backend-ci.yml @@ -1409,6 +1409,10 @@ jobs: env: DATABASE_URL: postgresql://postgres:postgres@localhost:5432/live_suite # pragma: allowlist secret ADMIN_DATABASE_URL: postgresql://regengine_sysadmin:admin-runtime-test-password@localhost:5432/live_suite # pragma: allowlist secret + # No object storage in this job. Partner uploads fail closed unless + # the explicit non-production bytea fallback is on, so opt in here + # and the CSV live tests keep exercising the runtime role. + REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK: "1" run: | repo_root="$(pwd)" export PYTHONPATH="${repo_root}:${repo_root}/services:${PYTHONPATH}" diff --git a/services/ingestion/tests/test_webhook_compat_1342.py b/services/ingestion/tests/test_webhook_compat_1342.py index 0e1172a8c..69056a952 100644 --- a/services/ingestion/tests/test_webhook_compat_1342.py +++ b/services/ingestion/tests/test_webhook_compat_1342.py @@ -752,6 +752,17 @@ def test_open_records_file_hash_size_rows_and_retains_bytes(self, monkeypatch): created: dict = {} + # Bytes go through persist_partner_upload. Pin the explicit local + # bytea fallback so the run's bytes land in fsma.manifest_blobs and + # the test never reaches a real bucket. + import app.remediation.audit_utils as audit_utils + import app.s3_utils as s3_utils + + monkeypatch.setenv("REGENGINE_ENV", "development") + monkeypatch.setenv("REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK", "1") + monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: False) + monkeypatch.setattr(audit_utils, "audit_event", lambda *a, **k: None) + class _Store: def __init__(self, db_session, dual_write): created["store_session"] = db_session