From 889cafaf207a2e7e495c64ccbac07d9faf5a5dce Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 8 Sep 2026 12:19:38 +0200 Subject: [PATCH] fix(connectors): retain exact CSV source evidence Release v0.1.27. Coordinated integrity review: GovOPlaN/govoplan-core#298. --- package.json | 2 +- pyproject.toml | 4 +- src/govoplan_connectors/backend/db/models.py | 2 + .../backend/german_documentation.py | 16 +--- .../backend/knowledge_search.py | 26 +----- src/govoplan_connectors/backend/manifest.py | 34 +++++++- .../d2a4c6e8f0b1_csv_source_evidence.py | 20 +++++ src/govoplan_connectors/backend/router.py | 38 ++++++++- src/govoplan_connectors/backend/schemas.py | 2 + .../backend/search_principal.py | 26 ++++++ .../backend/service_desk_search.py | 26 +----- .../backend/tabular_adapters.py | 46 +++-------- .../backend/tabular_sources.py | 80 ++++++++++--------- tests/test_migrations.py | 35 +++++++- tests/test_search_principal.py | 52 ++++++++++++ tests/test_tabular_sources.py | 56 ++++++++++++- webui/package.json | 2 +- 17 files changed, 322 insertions(+), 145 deletions(-) create mode 100755 src/govoplan_connectors/backend/migrations/versions/d2a4c6e8f0b1_csv_source_evidence.py create mode 100755 src/govoplan_connectors/backend/search_principal.py create mode 100755 tests/test_search_principal.py diff --git a/package.json b/package.json index 4f29316..a103f28 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@govoplan/connectors-webui", - "version": "0.1.26", + "version": "0.1.27", "private": true, "type": "module", "main": "webui/src/index.ts", diff --git a/pyproject.toml b/pyproject.toml index fe63b76..b8030cf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "govoplan-connectors" -version = "0.1.26" +version = "0.1.27" description = "Governed connector catalogue and tabular source capabilities for GovOPlaN." readme = "README.md" requires-python = ">=3.12" @@ -12,7 +12,7 @@ license = "AGPL-3.0-or-later" authors = [{ name = "GovOPlaN" }] dependencies = [ "defusedxml>=0.7,<1", - "govoplan-core>=0.1.33", + "govoplan-core>=0.1.46", "openpyxl>=3.1.5,<4", ] diff --git a/src/govoplan_connectors/backend/db/models.py b/src/govoplan_connectors/backend/db/models.py index e5d04cb..bed49df 100644 --- a/src/govoplan_connectors/backend/db/models.py +++ b/src/govoplan_connectors/backend/db/models.py @@ -47,6 +47,8 @@ class ConnectorTabularSource(Base, TimestampMixin): row_count: Mapped[int] = mapped_column(Integer, nullable=False) byte_count: Mapped[int] = mapped_column(Integer, nullable=False) metadata_: Mapped[dict[str, Any]] = mapped_column("metadata", JSON, default=dict, nullable=False) + # Original upload content is deliberately excluded from ordinary catalogue loads/DTOs. + csv_source_: Mapped[dict[str, Any] | None] = mapped_column("csv_source", JSON, nullable=True, deferred=True) created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) updated_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) diff --git a/src/govoplan_connectors/backend/german_documentation.py b/src/govoplan_connectors/backend/german_documentation.py index ffdd101..bdf4c0f 100644 --- a/src/govoplan_connectors/backend/german_documentation.py +++ b/src/govoplan_connectors/backend/german_documentation.py @@ -1,9 +1,8 @@ from __future__ import annotations -from dataclasses import replace from typing import Iterable -from govoplan_core.core.modules import DocumentationTopic +from govoplan_core.core.modules import DocumentationTopic, localize_documentation_topics as _localize_topics _TRANSLATIONS = { @@ -64,15 +63,4 @@ _TRANSLATIONS = { def localize_documentation_topics( topics: Iterable[DocumentationTopic], ) -> tuple[DocumentationTopic, ...]: - localized: list[DocumentationTopic] = [] - for topic in topics: - german = _TRANSLATIONS.get(topic.id) - if german is None: - localized.append(topic) - continue - translations = { - locale: dict(value) for locale, value in topic.translations.items() - } - translations["de"] = {**translations.get("de", {}), **german} - localized.append(replace(topic, translations=translations)) - return tuple(localized) + return _localize_topics(topics, locale="de", translations=_TRANSLATIONS) diff --git a/src/govoplan_connectors/backend/knowledge_search.py b/src/govoplan_connectors/backend/knowledge_search.py index 3ba4312..ddde53c 100644 --- a/src/govoplan_connectors/backend/knowledge_search.py +++ b/src/govoplan_connectors/backend/knowledge_search.py @@ -1,5 +1,7 @@ from __future__ import annotations +from govoplan_connectors.backend.search_principal import principal_acl_tokens as _principal_acl_tokens + from collections.abc import Mapping, Sequence from urllib.parse import quote @@ -217,30 +219,6 @@ def search_document( ) -def _principal_acl_tokens(principal: object) -> tuple[str, ...]: - values: list[str] = [] - for prefix, attribute in ( - ("account", "account_id"), - ("membership", "membership_id"), - ("identity", "identity_id"), - ): - value = getattr(principal, attribute, None) - if value: - values.append(f"{prefix}:{value}") - for prefix, attribute in ( - ("group", "group_ids"), - ("role", "role_ids"), - ("function", "function_assignment_ids"), - ("scope", "scopes"), - ): - values.extend( - f"{prefix}:{value}" - for value in getattr(principal, attribute, ()) - if value - ) - return tuple(dict.fromkeys(values))[:500] - - def _has_scope(principal: object, required: str) -> bool: check = getattr(principal, "has", None) if callable(check): diff --git a/src/govoplan_connectors/backend/manifest.py b/src/govoplan_connectors/backend/manifest.py index 0924d10..9686b58 100644 --- a/src/govoplan_connectors/backend/manifest.py +++ b/src/govoplan_connectors/backend/manifest.py @@ -126,7 +126,7 @@ from govoplan_connectors.backend.german_documentation import ( MODULE_ID = "connectors" -MODULE_VERSION = "0.1.26" +MODULE_VERSION = "0.1.27" TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0" DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0" SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0" @@ -967,6 +967,38 @@ manifest = ModuleManifest( ), ), documentation=localize_documentation_topics(( + DocumentationTopic( + id="connectors.csv-source-fidelity", + title="Preserving original CSV input", + summary="Choose explicit text or legacy type inference, and retrieve the verified original of new CSV snapshots.", + body=( + "CSV snapshot and managed-file creation accept csv_value_mode=text to preserve field whitespace, decimal digits, large identifiers, boolean-looking text and explicit empty records as strings. " + "Text mode rejects missing/extra fields and malformed quoting rather than dropping values. Headers remain normalized and blank physical lines are not table rows; the original upload preserves these lexical details. " + "Omitting csv_value_mode keeps legacy_typed API behavior for existing clients. Legacy inference can trim values, infer numbers/booleans and omit empty rows; choose text when exact values matter. " + "New CSV snapshots retain the original supplied text, delimiter, parser profile and UTF-8 SHA-256 separately from catalogue responses; csv_source metadata is reserved verified evidence, not a place to supply content. " + "GET /api/v1/connectors/tabular-sources/{source_ref}/original-csv requires current connectors:source:read or connectors:source:admin authority in the same tenant, verifies the checksum, audits the export without source content, and returns an uncached attachment. " + "The export preserves the text submitted to the API as UTF-8, not an earlier file encoding. It does not sanitize spreadsheet formulas: treat imported originals as untrusted data. " + "Original text and parsed rows each remain limited to 5 MB, with at most 10,000 parsed rows. Each snapshot can retain up to 5 MB of original text, plus serialization and metadata storage overhead. Source text follows the snapshot lifecycle; retired/deleted sources are not downloadable. " + "Managed files retain their original through the exact Files version and its access/lifecycle policy; text mode is pinned in source metadata and preserved on refresh. Existing snapshot fingerprints and historical rows are never rewritten. " + "Original text cannot be reconstructed for older snapshots: the endpoint reports it unavailable. Schema inference now shares one ordered, bounded-state mechanism with Datasources while preserving legacy provider type classifications." + ), + layer="always", documentation_types=("user", "admin"), audience=("user", "module_admin", "operator"), order=8, + translations={"de": { + "title": "Ursprüngliche CSV-Eingaben erhalten", + "summary": "Textwerte oder bisherige Typableitung ausdrücklich wählen und das geprüfte Original neuer CSV-Snapshots abrufen.", + "body": ( + "CSV-Snapshots und verwaltete Dateiquellen unterstützen csv_value_mode=text. Dieser Modus erhält Leerzeichen in Feldwerten, Dezimalstellen, große Kennungen, boolesch wirkenden Text und ausdrücklich leere Datensätze als Zeichenketten. " + "Fehlende oder zusätzliche Felder und fehlerhafte Anführungszeichen werden abgelehnt. Überschriften werden weiterhin normalisiert; vollständig leere physische Zeilen sind keine Tabellenzeilen. Das Original erhält auch diese Texteigenschaften. " + "Ohne csv_value_mode bleibt für bestehende API-Aufrufe legacy_typed aktiv. Dabei können Werte gekürzt, Zahlen/Wahrheitswerte abgeleitet und leere Zeilen ausgelassen werden. Für genaue Werte wählen Sie Text. " + "Neue CSV-Snapshots speichern den gelieferten Originaltext, Trennzeichen, Parserprofil und UTF-8-SHA-256 getrennt vom Katalog. csv_source-Metadaten sind reservierter geprüfter Nachweis. " + "GET /api/v1/connectors/tabular-sources/{source_ref}/original-csv benötigt aktuelle Rechte connectors:source:read oder connectors:source:admin im selben Mandanten, prüft die Prüfsumme, protokolliert den Export ohne Quellinhalt und liefert einen nicht zwischengespeicherten Download. " + "Exportiert wird der an die API übermittelte Text als UTF-8, nicht eine frühere Dateikodierung. Tabellenformeln werden nicht verändert; behandeln Sie Originaldateien als nicht vertrauenswürdige Daten. " + "Originaltext und verarbeitete Zeilen sind jeweils auf 5 MB begrenzt; höchstens 10.000 Zeilen werden angenommen. Pro Snapshot werden bis zu 5 MB Originaltext zuzüglich Speicher für Serialisierung und Metadaten aufbewahrt. Der Originaltext folgt dem Snapshot-Lebenszyklus; stillgelegte oder gelöschte Quellen sind nicht abrufbar. " + "Verwaltete Dateien behalten ihr Original in der genauen Files-Version mit deren Zugriffs- und Lebenszyklusregeln; der Textmodus bleibt bei Aktualisierungen erhalten. Bestehende Fingerprints und historische Zeilen werden nicht umgeschrieben. " + "Für ältere Snapshots kann das Original nicht rekonstruiert werden; der Abruf meldet es als nicht verfügbar. Die gemeinsame Schemaableitung erhält die bisherigen Typregeln und Spaltenreihenfolge." + ), + }}, + ), DocumentationTopic( id="connectors.data-subject-requests", title="Connector data-subject requests", diff --git a/src/govoplan_connectors/backend/migrations/versions/d2a4c6e8f0b1_csv_source_evidence.py b/src/govoplan_connectors/backend/migrations/versions/d2a4c6e8f0b1_csv_source_evidence.py new file mode 100755 index 0000000..a728adf --- /dev/null +++ b/src/govoplan_connectors/backend/migrations/versions/d2a4c6e8f0b1_csv_source_evidence.py @@ -0,0 +1,20 @@ +"""Retain original CSV source evidence for new durable snapshots.""" + +from alembic import op +import sqlalchemy as sa + +revision = "d2a4c6e8f0b1" +down_revision = "c0f1a2b3c4d5" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "connector_tabular_sources", sa.Column("csv_source", sa.JSON(), nullable=True) + ) + + +def downgrade() -> None: + with op.batch_alter_table("connector_tabular_sources") as batch: + batch.drop_column("csv_source") diff --git a/src/govoplan_connectors/backend/router.py b/src/govoplan_connectors/backend/router.py index 7983b55..42df6c3 100644 --- a/src/govoplan_connectors/backend/router.py +++ b/src/govoplan_connectors/backend/router.py @@ -10,6 +10,7 @@ from sqlalchemy.orm import Session from govoplan_core.audit.logging import audit_event from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope from govoplan_core.core.tabular_sources import ( + TabularCsvSource, TabularReadRequest, TabularSnapshotInput, TabularSource, @@ -539,7 +540,7 @@ def api_create_tabular_snapshot( rows = ( tuple(payload.rows or ()) if payload.format == "json" - else parse_csv_snapshot(payload.csv_text or "", delimiter=payload.delimiter) + else parse_csv_snapshot(payload.csv_text or "", delimiter=payload.delimiter, value_mode=payload.csv_value_mode) ) source = provider.create_snapshot( session, @@ -550,6 +551,11 @@ def api_create_tabular_snapshot( description=payload.description, rows=rows, metadata={"import_format": payload.format}, + csv_source=(TabularCsvSource( + text=payload.csv_text or "", + delimiter=payload.delimiter, + value_mode=payload.csv_value_mode, + ) if payload.format == "csv" else None), ), ) except TabularSourceError as exc: @@ -573,6 +579,35 @@ def api_create_tabular_snapshot( return _source_response(source) +@router.get("/tabular-sources/{source_ref}/original-csv") +def api_original_tabular_csv( + source_ref: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> Response: + _require_any_scope(principal, READ_SCOPE, ADMIN_SCOPE) + try: + text = provider.original_csv(session, principal, source_ref=source_ref) + except TabularSourceError as exc: + raise _http_error(exc) from exc + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=getattr(principal.user, "id", None), + api_key_id=principal.api_key_id, + action="connectors.original_csv.exported", + object_type="connector_tabular_source", + object_id=source_ref, + details={"sha256": hashlib.sha256(text.encode("utf-8")).hexdigest()}, + ) + session.commit() + return Response( + content=text.encode("utf-8"), + media_type="text/csv; charset=utf-8", + headers={"Content-Disposition": 'attachment; filename="original.csv"', "Cache-Control": "no-store"}, + ) + + @router.post( "/tabular-sources/files", response_model=TabularSourceResponse, @@ -595,6 +630,7 @@ def api_create_managed_file_source( file_version_id=payload.file_version_id, delimiter=payload.delimiter, sheet_name=payload.sheet_name, + csv_value_mode=payload.csv_value_mode, ) except TabularSourceError as exc: raise _http_error(exc) from exc diff --git a/src/govoplan_connectors/backend/schemas.py b/src/govoplan_connectors/backend/schemas.py index f9e1b93..0885fd2 100644 --- a/src/govoplan_connectors/backend/schemas.py +++ b/src/govoplan_connectors/backend/schemas.py @@ -90,6 +90,7 @@ class SnapshotCreateRequest(BaseModel): format: Literal["json", "csv"] = "json" rows: list[dict[str, Any]] | None = Field(default=None, max_length=10_000) csv_text: str | None = Field(default=None, max_length=5_000_000) + csv_value_mode: Literal["legacy_typed", "text"] = "legacy_typed" delimiter: Literal[",", ";", "\t", "|"] = "," @model_validator(mode="after") @@ -117,6 +118,7 @@ class ManagedFileSourceCreateRequest(BaseModel): file_version_id: str | None = Field(default=None, min_length=1, max_length=36) delimiter: Literal[",", ";", "\t", "|"] = "," sheet_name: str | None = Field(default=None, min_length=1, max_length=255) + csv_value_mode: Literal["legacy_typed", "text"] = "legacy_typed" class SqlSourceCreateRequest(BaseModel): diff --git a/src/govoplan_connectors/backend/search_principal.py b/src/govoplan_connectors/backend/search_principal.py new file mode 100755 index 0000000..8b2698b --- /dev/null +++ b/src/govoplan_connectors/backend/search_principal.py @@ -0,0 +1,26 @@ +"""Shared connector-search ACL token projection; owner authorization stays local.""" + + +def principal_acl_tokens(principal: object) -> tuple[str, ...]: + # Keep legacy first-seen ordering and the exact 500-token authorization cap. + values: dict[str, None] = {} + for prefix, attribute in ( + ("account", "account_id"), + ("membership", "membership_id"), + ("identity", "identity_id"), + ): + value = getattr(principal, attribute, None) + if value: + values[f"{prefix}:{value}"] = None + for prefix, attribute in ( + ("group", "group_ids"), + ("role", "role_ids"), + ("function", "function_assignment_ids"), + ("scope", "scopes"), + ): + for value in getattr(principal, attribute, ()): + if value: + values[f"{prefix}:{value}"] = None + if len(values) >= 500: + return tuple(values) + return tuple(values) diff --git a/src/govoplan_connectors/backend/service_desk_search.py b/src/govoplan_connectors/backend/service_desk_search.py index 2c6c307..380f3b2 100644 --- a/src/govoplan_connectors/backend/service_desk_search.py +++ b/src/govoplan_connectors/backend/service_desk_search.py @@ -1,5 +1,7 @@ from __future__ import annotations +from govoplan_connectors.backend.search_principal import principal_acl_tokens as _principal_acl_tokens + from collections.abc import Mapping, Sequence from urllib.parse import quote @@ -288,30 +290,6 @@ def search_document( ) -def _principal_acl_tokens(principal: object) -> tuple[str, ...]: - values: list[str] = [] - for prefix, attribute in ( - ("account", "account_id"), - ("membership", "membership_id"), - ("identity", "identity_id"), - ): - value = getattr(principal, attribute, None) - if value: - values.append(f"{prefix}:{value}") - for prefix, attribute in ( - ("group", "group_ids"), - ("role", "role_ids"), - ("function", "function_assignment_ids"), - ("scope", "scopes"), - ): - values.extend( - f"{prefix}:{value}" - for value in getattr(principal, attribute, ()) - if value - ) - return tuple(dict.fromkeys(values))[:500] - - def _has_scope(principal: object, required: str) -> bool: check = getattr(principal, "has", None) if callable(check): diff --git a/src/govoplan_connectors/backend/tabular_adapters.py b/src/govoplan_connectors/backend/tabular_adapters.py index 3bf0fe7..a9ba40b 100644 --- a/src/govoplan_connectors/backend/tabular_adapters.py +++ b/src/govoplan_connectors/backend/tabular_adapters.py @@ -53,6 +53,9 @@ from govoplan_core.core.tabular_sources import ( TabularSourceUnavailableError, TabularSourceValidationError, parse_tabular_csv, + CsvValueMode, + infer_tabular_schema as _infer_schema, + tabular_type_name, ) from govoplan_core.security.credential_envelopes import ( CredentialAccessContext, @@ -125,6 +128,7 @@ class ManagedFileTabularAdapter: file_version_id: str | None, delimiter: str = ",", sheet_name: str | None = None, + csv_value_mode: CsvValueMode = "legacy_typed", ) -> TabularOriginInspection: provider = managed_tabular_file_provider(self._registry) if provider is None: @@ -163,6 +167,7 @@ class ManagedFileTabularAdapter: content_type=content.file.content_type, delimiter=delimiter, sheet_name=sheet_name, + csv_value_mode=csv_value_mode, ) schema = infer_tabular_schema(rows) fingerprint = origin_fingerprint( @@ -174,6 +179,7 @@ class ManagedFileTabularAdapter: content.file.sha256, resolved_sheet or "", delimiter, + *(("csv_text_values",) if csv_value_mode == "text" else ()), ), ) try: @@ -227,6 +233,7 @@ class ManagedFileTabularAdapter: "content_type": content.file.content_type, "format": "xlsx" if _is_xlsx(content.file.filename, content.file.content_type) else "csv", "delimiter": delimiter, + "csv_value_mode": csv_value_mode, "sheet_name": resolved_sheet, }, health=health, @@ -263,6 +270,7 @@ class ManagedFileTabularAdapter: file_version_id=_required_metadata(metadata, "file_version_id"), delimiter=str(metadata.get("delimiter") or ","), sheet_name=_optional_text(metadata.get("sheet_name")), + csv_value_mode=str(metadata.get("csv_value_mode") or "legacy_typed"), ) expected_sha256 = _required_metadata(metadata, "file_sha256") if inspection.metadata.get("file_sha256") != expected_sha256: @@ -567,6 +575,7 @@ def parse_managed_tabular_content( content_type: str | None, delimiter: str, sheet_name: str | None, + csv_value_mode: CsvValueMode = "legacy_typed", ) -> tuple[tuple[Mapping[str, object], ...], str | None]: if len(payload) > MAX_FILE_BYTES: raise TabularSourceValidationError( @@ -581,7 +590,7 @@ def parse_managed_tabular_content( "Managed CSV files must use UTF-8 encoding." ) from exc return ( - tuple(parse_tabular_csv(text, delimiter=delimiter, max_rows=MAX_FILE_ROWS)), + tuple(parse_tabular_csv(text, delimiter=delimiter, max_rows=MAX_FILE_ROWS, max_bytes=MAX_FILE_BYTES, value_mode=csv_value_mode)), None, ) @@ -809,26 +818,7 @@ def _xlsx_headers(values: Sequence[object]) -> tuple[str, ...]: def infer_tabular_schema( rows: Sequence[Mapping[str, object]], ) -> tuple[TabularColumn, ...]: - names: list[str] = [] - for row in rows: - for name in row: - if name not in names: - names.append(name) - result: list[TabularColumn] = [] - for name in names: - values = [row.get(name) for row in rows] - concrete = [value for value in values if value is not None] - data_type = _type_name(concrete[0]) if concrete else "unknown" - if any(_type_name(value) != data_type for value in concrete[1:]): - data_type = "mixed" - result.append( - TabularColumn( - name=name, - data_type=data_type, - nullable=len(concrete) != len(values), - ) - ) - return tuple(result) + return _infer_schema(rows, type_name=_type_name) def origin_fingerprint( @@ -951,19 +941,7 @@ def _json_value(value: object) -> object: def _type_name(value: object) -> str: - if isinstance(value, bool): - return "boolean" - if isinstance(value, int): - return "integer" - if isinstance(value, (float, Decimal)): - return "number" - if isinstance(value, str): - return "string" - if isinstance(value, list): - return "array" - if isinstance(value, dict): - return "object" - return type(value).__name__.casefold() + return tabular_type_name(value, casefold_unknown=True) __all__ = [ diff --git a/src/govoplan_connectors/backend/tabular_sources.py b/src/govoplan_connectors/backend/tabular_sources.py index bc5f791..bb36d14 100644 --- a/src/govoplan_connectors/backend/tabular_sources.py +++ b/src/govoplan_connectors/backend/tabular_sources.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal, has_scope from govoplan_core.core.tabular_sources import ( + CsvValueMode, TabularColumn, TabularPreviewDiagnostic, TabularPushdown, @@ -26,6 +27,12 @@ from govoplan_core.core.tabular_sources import ( TabularSourceUnavailableError, TabularSourceValidationError, parse_tabular_csv, + csv_source_payload, + csv_source_summary, + csv_projection_matches, + verified_csv_source_text, + infer_tabular_schema, + tabular_type_name as _type_name, ) from govoplan_core.core.runtime import get_registry from govoplan_core.db.base import utcnow @@ -281,6 +288,7 @@ class SqlTabularSourceProvider: description: str | None = None, delimiter: str = ",", sheet_name: str | None = None, + csv_value_mode: CsvValueMode = "legacy_typed", ) -> TabularSource: db, api_principal = _context(session, principal, WRITE_SCOPE) inspection = self._file_adapter().inspect( @@ -290,6 +298,7 @@ class SqlTabularSourceProvider: file_version_id=file_version_id, delimiter=delimiter, sheet_name=sheet_name, + csv_value_mode=csv_value_mode, ) return self._create_origin( db, @@ -353,6 +362,7 @@ class SqlTabularSourceProvider: file_version_id=None, delimiter=str(item.metadata_.get("delimiter") or ","), sheet_name=_clean_optional(item.metadata_.get("sheet_name")), + csv_value_mode=str(item.metadata_.get("csv_value_mode") or "legacy_typed"), ) elif item.provider == "postgresql": inspection = self._sql_adapter.inspect( @@ -456,6 +466,20 @@ class SqlTabularSourceProvider: f"Snapshots are limited to {MAX_SNAPSHOT_ROWS:,} rows." ) rows = [_json_row(row) for row in snapshot.rows] + csv_payload = None + if snapshot.csv_source is not None: + if "csv_source" in snapshot.metadata: + raise TabularSourceValidationError("csv_source metadata is reserved for verified CSV import evidence.") + if snapshot.csv_source.parser_profile != "core.csv.v1": + raise TabularSourceValidationError("Unsupported original CSV parser profile.") + csv_payload = csv_source_payload(snapshot.csv_source, max_bytes=MAX_SNAPSHOT_BYTES) + expected = parse_csv_snapshot( + snapshot.csv_source.text, + delimiter=snapshot.csv_source.delimiter, + value_mode=snapshot.csv_source.value_mode, + ) + if not csv_projection_matches(expected, rows): + raise TabularSourceValidationError("Snapshot rows do not match their original CSV source and parsing mode.") encoded = json.dumps(rows, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") if len(encoded) > MAX_SNAPSHOT_BYTES: raise TabularSourceValidationError( @@ -487,7 +511,11 @@ class SqlTabularSourceProvider: fingerprint=fingerprint, row_count=len(rows), byte_count=len(encoded), - metadata_=dict(snapshot.metadata), + metadata_={ + **dict(snapshot.metadata), + **({"csv_source": csv_source_summary(csv_payload)} if csv_payload is not None else {}), + }, + csv_source_=csv_payload, created_by=actor_id, updated_by=actor_id, ) @@ -497,6 +525,15 @@ class SqlTabularSourceProvider: db.flush() return _source_dto(item) + def original_csv(self, session: object, principal: object, *, source_ref: str) -> str: + db, api_principal = _context(session, principal, READ_SCOPE) + item = _source_record(db, tenant_id=api_principal.tenant_id, source_ref=source_ref) + if item is None or item.status != "active": + raise TabularSourceNotFoundError("Tabular source not found.") + if not item.csv_source_: + raise TabularSourceNotFoundError("Original CSV was not retained for this source; historical typed snapshots cannot reconstruct it.") + return verified_csv_source_text(item.csv_source_, expected_summary=item.metadata_.get("csv_source") or {}) + def delete_snapshot( self, session: object, @@ -519,35 +556,18 @@ class SqlTabularSourceProvider: return _source_dto(item) -def parse_csv_snapshot(csv_text: str, *, delimiter: str) -> tuple[Mapping[str, object], ...]: +def parse_csv_snapshot(csv_text: str, *, delimiter: str, value_mode: CsvValueMode = "legacy_typed") -> tuple[Mapping[str, object], ...]: return parse_tabular_csv( csv_text, delimiter=delimiter, max_rows=MAX_SNAPSHOT_ROWS, + max_bytes=MAX_SNAPSHOT_BYTES, + value_mode=value_mode, ) def infer_schema(rows: Sequence[Mapping[str, object]]) -> tuple[TabularColumn, ...]: - names: list[str] = [] - for row in rows: - for name in row: - if name not in names: - names.append(name) - result: list[TabularColumn] = [] - for name in names: - values = [row.get(name) for row in rows] - concrete = [value for value in values if value is not None] - data_type = _type_name(concrete[0]) if concrete else "unknown" - if any(_type_name(value) != data_type for value in concrete[1:]): - data_type = "mixed" - result.append( - TabularColumn( - name=name, - data_type=data_type, - nullable=len(concrete) != len(values), - ) - ) - return tuple(result) + return infer_tabular_schema(rows, type_name=_type_name) def snapshot_fingerprint( @@ -717,22 +737,6 @@ def _unsupported_json(value: object) -> object: raise TypeError(f"{type(value).__name__} is not JSON serializable") -def _type_name(value: object) -> str: - if isinstance(value, bool): - return "boolean" - if isinstance(value, int): - return "integer" - if isinstance(value, (float, Decimal)): - return "number" - if isinstance(value, str): - return "string" - if isinstance(value, list): - return "array" - if isinstance(value, dict): - return "object" - return type(value).__name__.lower() - - def _column_payload(column: TabularColumn) -> dict[str, object]: return { "name": column.name, diff --git a/tests/test_migrations.py b/tests/test_migrations.py index e5343d8..39b26e2 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -2,16 +2,45 @@ from __future__ import annotations import tempfile import unittest +from datetime import UTC, datetime from pathlib import Path +from alembic import command from alembic.runtime.migration import MigrationContext -from sqlalchemy import create_engine, inspect +from sqlalchemy import MetaData, Table, create_engine, inspect, select from govoplan_connectors.backend.manifest import get_manifest -from govoplan_core.db.migrations import migrate_database +from govoplan_core.db.migrations import alembic_config, migrate_database class ConnectorsMigrationTests(unittest.TestCase): + def test_csv_evidence_upgrade_preserves_legacy_snapshot_without_inventing_source(self) -> None: + with tempfile.TemporaryDirectory(prefix="govoplan-connectors-csv-migration-") as directory: + url = f"sqlite:///{Path(directory) / 'connectors.db'}" + config = alembic_config(database_url=url, enabled_modules=("connectors",), manifest_factories=(get_manifest,)) + command.upgrade(config, "c0f1a2b3c4d5") + engine = create_engine(url) + try: + table = Table("connector_tabular_sources", MetaData(), autoload_with=engine) + now = datetime.now(UTC) + with engine.begin() as connection: + connection.execute(table.insert().values( + id="legacy-csv", tenant_id="tenant-1", provider="snapshot", + source_name="legacy", name="Legacy", status="active", schema_version=1, + schema=[{"name": "id", "data_type": "integer", "nullable": False}], + rows=[{"id": 1}], fingerprint="a" * 64, row_count=1, byte_count=10, + metadata={"original_label": "CSV"}, created_at=now, updated_at=now, + )) + before = dict(connection.execute(select(table)).mappings().one()) + command.upgrade(config, "d2a4c6e8f0b1") + upgraded = Table("connector_tabular_sources", MetaData(), autoload_with=engine) + with engine.connect() as connection: + after = dict(connection.execute(select(upgraded)).mappings().one()) + self.assertIsNone(after.pop("csv_source")) + self.assertEqual(before, after) + finally: + engine.dispose() + def test_baseline_creates_connector_tables_and_head(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-connectors-migration-") as directory: url = f"sqlite:///{Path(directory) / 'connectors.db'}" @@ -24,7 +53,7 @@ class ConnectorsMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "c0f1a2b3c4d5", + "d2a4c6e8f0b1", set(MigrationContext.configure(connection).get_current_heads()), ) self.assertTrue( diff --git a/tests/test_search_principal.py b/tests/test_search_principal.py new file mode 100755 index 0000000..bd25e16 --- /dev/null +++ b/tests/test_search_principal.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace + +from govoplan_connectors.backend.search_principal import principal_acl_tokens + + +class SearchPrincipalTests(unittest.TestCase): + def test_legacy_first_seen_projection_and_cap_are_unchanged(self) -> None: + principal = SimpleNamespace( + account_id=" actor ", + membership_id="m", + identity_id="i", + group_ids=("g", "", "g", "2"), + role_ids=("r", "r"), + function_assignment_ids=("f",), + scopes=tuple(f"scope-{i}" for i in range(700)), + ) + legacy = [] + for prefix, attribute in ( + ("account", "account_id"), + ("membership", "membership_id"), + ("identity", "identity_id"), + ): + value = getattr(principal, attribute, None) + if value: + legacy.append(f"{prefix}:{value}") + for prefix, attribute in ( + ("group", "group_ids"), + ("role", "role_ids"), + ("function", "function_assignment_ids"), + ("scope", "scopes"), + ): + legacy.extend( + f"{prefix}:{value}" + for value in getattr(principal, attribute, ()) + if value + ) + expected = tuple(dict.fromkeys(legacy))[:500] + self.assertEqual(expected, principal_acl_tokens(principal)) + self.assertEqual(500, len(expected)) + self.assertEqual((), principal_acl_tokens(object())) + + def test_projection_stops_consuming_at_authorization_cap(self) -> None: + def bounded_scopes(): + yield from (str(i) for i in range(500)) + raise AssertionError("ACL projection scanned beyond its effective cap") + + self.assertEqual( + 500, len(principal_acl_tokens(SimpleNamespace(scopes=bounded_scopes()))) + ) diff --git a/tests/test_tabular_sources.py b/tests/test_tabular_sources.py index f5725eb..d7e3d67 100644 --- a/tests/test_tabular_sources.py +++ b/tests/test_tabular_sources.py @@ -1,23 +1,27 @@ from __future__ import annotations import unittest +from unittest.mock import patch +from dataclasses import replace from fastapi import HTTPException -from sqlalchemy import create_engine +from sqlalchemy import create_engine, inspect from sqlalchemy.orm import sessionmaker from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.tabular_sources import ( + TabularCsvSource, TabularReadRequest, TabularSnapshotInput, TabularSourceAccessError, TabularSourceUnavailableError, TabularSourceValidationError, + TabularSourceNotFoundError, ) from govoplan_core.db.base import Base from govoplan_connectors.backend.db.models import ConnectorTabularSource -from govoplan_connectors.backend.router import api_create_tabular_snapshot +from govoplan_connectors.backend.router import api_create_tabular_snapshot, api_original_tabular_csv from govoplan_connectors.backend.schemas import SnapshotCreateRequest from govoplan_connectors.backend.tabular_sources import ( READ_SCOPE, @@ -263,6 +267,54 @@ class ConnectorsTabularSourceTests(unittest.TestCase): self.assertEqual(422, raised.exception.status_code) + def test_original_csv_round_trip_is_private_bounded_and_not_catalogue_content(self) -> None: + text = '\ufeffid,value\r\n9007199254740993," keep me "\r\ntrue,0.123456789012345678901234567890\r\n" ",""\r\n' + payload = SnapshotCreateRequest(name="CSV", source_name="csv", format="csv", csv_text=text) + with patch("govoplan_connectors.backend.router.audit_event"): + created = api_create_tabular_snapshot(payload, session=self.session, principal=principal()) + self.session.expunge_all() + listed = self.provider.list_sources(self.session, principal()) + self.assertNotIn("text", listed[0].metadata["csv_source"]) + self.assertEqual("legacy_typed", listed[0].metadata["csv_source"]["value_mode"]) + record = self.session.get(ConnectorTabularSource, created.ref.split(":", 1)[1]) + self.assertIn("csv_source_", inspect(record).unloaded) + with patch("govoplan_connectors.backend.router.audit_event") as audit: + response = api_original_tabular_csv(created.ref, session=self.session, principal=principal()) + self.assertEqual("connectors.original_csv.exported", audit.call_args.kwargs["action"]) + self.assertEqual({"sha256"}, set(audit.call_args.kwargs["details"])) + with patch("govoplan_connectors.backend.router.provider.original_csv") as read, self.assertRaises(HTTPException) as denied: + api_original_tabular_csv(created.ref, session=self.session, principal=principal(scopes=())) + self.assertEqual(403, denied.exception.status_code) + read.assert_not_called() + self.assertEqual(text.encode("utf-8"), response.body) + self.assertEqual("no-store", response.headers["cache-control"]) + self.assertIn("attachment", response.headers["content-disposition"]) + with self.assertRaises(TabularSourceNotFoundError): + self.provider.original_csv(self.session, principal("tenant-2"), source_ref=created.ref) + with self.assertRaises(TabularSourceAccessError): + self.provider.original_csv(self.session, principal(scopes=()), source_ref=created.ref) + record.csv_source_ = {**record.csv_source_, "text": "tampered"} + self.session.flush() + with self.assertRaises(TabularSourceUnavailableError): + self.provider.original_csv(self.session, principal(), source_ref=created.ref) + + def test_text_snapshot_matches_exact_projection_and_rejects_inconsistent_evidence(self) -> None: + source = TabularCsvSource(text='value\n" "\n0.123456789012345678901234567890\n', value_mode="text") + snapshot = TabularSnapshotInput(name="Text", source_name="text", rows=parse_csv_snapshot(source.text, delimiter=",", value_mode="text"), csv_source=source) + with self.assertRaises(TabularSourceValidationError): + self.provider.create_snapshot(self.session, principal(), snapshot=replace(snapshot, rows=({"value": "changed"},))) + created = self.provider.create_snapshot(self.session, principal(), snapshot=snapshot) + self.assertEqual(2, created.row_count) + self.assertEqual(source.text, self.provider.original_csv(self.session, principal(), source_ref=created.ref)) + legacy = self.provider.create_snapshot(self.session, principal(), snapshot=TabularSnapshotInput(name="Old", source_name="old", rows=({"id": 1},))) + with self.assertRaises(TabularSourceNotFoundError): + self.provider.original_csv(self.session, principal(), source_ref=legacy.ref) + + def test_original_csv_projection_binding_rejects_equal_but_different_types(self) -> None: + for text, value in (("value\ntrue\n", 1), ("value\n1\n", True), ("value\n1\n", 1.0)): + with self.subTest(text=text, value=value), self.assertRaises(TabularSourceValidationError): + self.provider.create_snapshot(self.session, principal(), snapshot=TabularSnapshotInput(name="Invalid", source_name="invalid", rows=({"value": value},), csv_source=TabularCsvSource(text=text))) + if __name__ == "__main__": unittest.main() diff --git a/webui/package.json b/webui/package.json index e1f2528..16ab5d7 100644 --- a/webui/package.json +++ b/webui/package.json @@ -1,6 +1,6 @@ { "name": "@govoplan/connectors-webui", - "version": "0.1.26", + "version": "0.1.27", "private": true, "type": "module", "main": "src/index.ts",