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/app/partner_file_storage.py b/services/ingestion/app/partner_file_storage.py new file mode 100644 index 000000000..6f2e1a699 --- /dev/null +++ b/services/ingestion/app/partner_file_storage.py @@ -0,0 +1,246 @@ +"""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 os +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" + +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._-]+") + + +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 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``. + """ + 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] = None + + 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, + }, + ) + 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 + 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..491c3c9ec --- /dev/null +++ b/services/ingestion/tests/test_partner_file_storage_cutover.py @@ -0,0 +1,209 @@ +"""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_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) + + 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_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" + + +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" 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