fix(connectors): retain exact CSV source evidence

Release v0.1.27. Coordinated integrity review: GovOPlaN/govoplan-core#298.
This commit is contained in:
2026-09-08 12:19:38 +02:00
parent be16f8218e
commit 2c7f4ca9ed
17 changed files with 322 additions and 145 deletions
+1 -1
View File
@@ -1,6 +1,6 @@
{ {
"name": "@govoplan/connectors-webui", "name": "@govoplan/connectors-webui",
"version": "0.1.26", "version": "0.1.27",
"private": true, "private": true,
"type": "module", "type": "module",
"main": "webui/src/index.ts", "main": "webui/src/index.ts",
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "govoplan-connectors" name = "govoplan-connectors"
version = "0.1.26" version = "0.1.27"
description = "Governed connector catalogue and tabular source capabilities for GovOPlaN." description = "Governed connector catalogue and tabular source capabilities for GovOPlaN."
readme = "README.md" readme = "README.md"
requires-python = ">=3.12" requires-python = ">=3.12"
@@ -12,7 +12,7 @@ license = "AGPL-3.0-or-later"
authors = [{ name = "GovOPlaN" }] authors = [{ name = "GovOPlaN" }]
dependencies = [ dependencies = [
"defusedxml>=0.7,<1", "defusedxml>=0.7,<1",
"govoplan-core>=0.1.33", "govoplan-core>=0.1.46",
"openpyxl>=3.1.5,<4", "openpyxl>=3.1.5,<4",
] ]
@@ -47,6 +47,8 @@ class ConnectorTabularSource(Base, TimestampMixin):
row_count: Mapped[int] = mapped_column(Integer, nullable=False) row_count: Mapped[int] = mapped_column(Integer, nullable=False)
byte_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) 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) 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) 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) deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
@@ -1,9 +1,8 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import replace
from typing import Iterable 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 = { _TRANSLATIONS = {
@@ -64,15 +63,4 @@ _TRANSLATIONS = {
def localize_documentation_topics( def localize_documentation_topics(
topics: Iterable[DocumentationTopic], topics: Iterable[DocumentationTopic],
) -> tuple[DocumentationTopic, ...]: ) -> tuple[DocumentationTopic, ...]:
localized: list[DocumentationTopic] = [] return _localize_topics(topics, locale="de", translations=_TRANSLATIONS)
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)
@@ -1,5 +1,7 @@
from __future__ import annotations 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 collections.abc import Mapping, Sequence
from urllib.parse import quote 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: def _has_scope(principal: object, required: str) -> bool:
check = getattr(principal, "has", None) check = getattr(principal, "has", None)
if callable(check): if callable(check):
+33 -1
View File
@@ -126,7 +126,7 @@ from govoplan_connectors.backend.german_documentation import (
MODULE_ID = "connectors" MODULE_ID = "connectors"
MODULE_VERSION = "0.1.26" MODULE_VERSION = "0.1.27"
TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0" TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0"
DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0" DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0"
SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0" SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0"
@@ -967,6 +967,38 @@ manifest = ModuleManifest(
), ),
), ),
documentation=localize_documentation_topics(( 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( DocumentationTopic(
id="connectors.data-subject-requests", id="connectors.data-subject-requests",
title="Connector data-subject requests", title="Connector data-subject requests",
@@ -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")
+37 -1
View File
@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session
from govoplan_core.audit.logging import audit_event from govoplan_core.audit.logging import audit_event
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
from govoplan_core.core.tabular_sources import ( from govoplan_core.core.tabular_sources import (
TabularCsvSource,
TabularReadRequest, TabularReadRequest,
TabularSnapshotInput, TabularSnapshotInput,
TabularSource, TabularSource,
@@ -539,7 +540,7 @@ def api_create_tabular_snapshot(
rows = ( rows = (
tuple(payload.rows or ()) tuple(payload.rows or ())
if payload.format == "json" 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( source = provider.create_snapshot(
session, session,
@@ -550,6 +551,11 @@ def api_create_tabular_snapshot(
description=payload.description, description=payload.description,
rows=rows, rows=rows,
metadata={"import_format": payload.format}, 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: except TabularSourceError as exc:
@@ -573,6 +579,35 @@ def api_create_tabular_snapshot(
return _source_response(source) 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( @router.post(
"/tabular-sources/files", "/tabular-sources/files",
response_model=TabularSourceResponse, response_model=TabularSourceResponse,
@@ -595,6 +630,7 @@ def api_create_managed_file_source(
file_version_id=payload.file_version_id, file_version_id=payload.file_version_id,
delimiter=payload.delimiter, delimiter=payload.delimiter,
sheet_name=payload.sheet_name, sheet_name=payload.sheet_name,
csv_value_mode=payload.csv_value_mode,
) )
except TabularSourceError as exc: except TabularSourceError as exc:
raise _http_error(exc) from exc raise _http_error(exc) from exc
@@ -90,6 +90,7 @@ class SnapshotCreateRequest(BaseModel):
format: Literal["json", "csv"] = "json" format: Literal["json", "csv"] = "json"
rows: list[dict[str, Any]] | None = Field(default=None, max_length=10_000) 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_text: str | None = Field(default=None, max_length=5_000_000)
csv_value_mode: Literal["legacy_typed", "text"] = "legacy_typed"
delimiter: Literal[",", ";", "\t", "|"] = "," delimiter: Literal[",", ";", "\t", "|"] = ","
@model_validator(mode="after") @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) file_version_id: str | None = Field(default=None, min_length=1, max_length=36)
delimiter: Literal[",", ";", "\t", "|"] = "," delimiter: Literal[",", ";", "\t", "|"] = ","
sheet_name: str | None = Field(default=None, min_length=1, max_length=255) 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): class SqlSourceCreateRequest(BaseModel):
+26
View File
@@ -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)
@@ -1,5 +1,7 @@
from __future__ import annotations 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 collections.abc import Mapping, Sequence
from urllib.parse import quote 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: def _has_scope(principal: object, required: str) -> bool:
check = getattr(principal, "has", None) check = getattr(principal, "has", None)
if callable(check): if callable(check):
@@ -53,6 +53,9 @@ from govoplan_core.core.tabular_sources import (
TabularSourceUnavailableError, TabularSourceUnavailableError,
TabularSourceValidationError, TabularSourceValidationError,
parse_tabular_csv, parse_tabular_csv,
CsvValueMode,
infer_tabular_schema as _infer_schema,
tabular_type_name,
) )
from govoplan_core.security.credential_envelopes import ( from govoplan_core.security.credential_envelopes import (
CredentialAccessContext, CredentialAccessContext,
@@ -125,6 +128,7 @@ class ManagedFileTabularAdapter:
file_version_id: str | None, file_version_id: str | None,
delimiter: str = ",", delimiter: str = ",",
sheet_name: str | None = None, sheet_name: str | None = None,
csv_value_mode: CsvValueMode = "legacy_typed",
) -> TabularOriginInspection: ) -> TabularOriginInspection:
provider = managed_tabular_file_provider(self._registry) provider = managed_tabular_file_provider(self._registry)
if provider is None: if provider is None:
@@ -163,6 +167,7 @@ class ManagedFileTabularAdapter:
content_type=content.file.content_type, content_type=content.file.content_type,
delimiter=delimiter, delimiter=delimiter,
sheet_name=sheet_name, sheet_name=sheet_name,
csv_value_mode=csv_value_mode,
) )
schema = infer_tabular_schema(rows) schema = infer_tabular_schema(rows)
fingerprint = origin_fingerprint( fingerprint = origin_fingerprint(
@@ -174,6 +179,7 @@ class ManagedFileTabularAdapter:
content.file.sha256, content.file.sha256,
resolved_sheet or "", resolved_sheet or "",
delimiter, delimiter,
*(("csv_text_values",) if csv_value_mode == "text" else ()),
), ),
) )
try: try:
@@ -227,6 +233,7 @@ class ManagedFileTabularAdapter:
"content_type": content.file.content_type, "content_type": content.file.content_type,
"format": "xlsx" if _is_xlsx(content.file.filename, content.file.content_type) else "csv", "format": "xlsx" if _is_xlsx(content.file.filename, content.file.content_type) else "csv",
"delimiter": delimiter, "delimiter": delimiter,
"csv_value_mode": csv_value_mode,
"sheet_name": resolved_sheet, "sheet_name": resolved_sheet,
}, },
health=health, health=health,
@@ -263,6 +270,7 @@ class ManagedFileTabularAdapter:
file_version_id=_required_metadata(metadata, "file_version_id"), file_version_id=_required_metadata(metadata, "file_version_id"),
delimiter=str(metadata.get("delimiter") or ","), delimiter=str(metadata.get("delimiter") or ","),
sheet_name=_optional_text(metadata.get("sheet_name")), 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") expected_sha256 = _required_metadata(metadata, "file_sha256")
if inspection.metadata.get("file_sha256") != expected_sha256: if inspection.metadata.get("file_sha256") != expected_sha256:
@@ -567,6 +575,7 @@ def parse_managed_tabular_content(
content_type: str | None, content_type: str | None,
delimiter: str, delimiter: str,
sheet_name: str | None, sheet_name: str | None,
csv_value_mode: CsvValueMode = "legacy_typed",
) -> tuple[tuple[Mapping[str, object], ...], str | None]: ) -> tuple[tuple[Mapping[str, object], ...], str | None]:
if len(payload) > MAX_FILE_BYTES: if len(payload) > MAX_FILE_BYTES:
raise TabularSourceValidationError( raise TabularSourceValidationError(
@@ -581,7 +590,7 @@ def parse_managed_tabular_content(
"Managed CSV files must use UTF-8 encoding." "Managed CSV files must use UTF-8 encoding."
) from exc ) from exc
return ( 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, None,
) )
@@ -809,26 +818,7 @@ def _xlsx_headers(values: Sequence[object]) -> tuple[str, ...]:
def infer_tabular_schema( def infer_tabular_schema(
rows: Sequence[Mapping[str, object]], rows: Sequence[Mapping[str, object]],
) -> tuple[TabularColumn, ...]: ) -> tuple[TabularColumn, ...]:
names: list[str] = [] return _infer_schema(rows, type_name=_type_name)
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)
def origin_fingerprint( def origin_fingerprint(
@@ -951,19 +941,7 @@ def _json_value(value: object) -> object:
def _type_name(value: object) -> str: def _type_name(value: object) -> str:
if isinstance(value, bool): return tabular_type_name(value, casefold_unknown=True)
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()
__all__ = [ __all__ = [
@@ -13,6 +13,7 @@ from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal, has_scope from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.core.tabular_sources import ( from govoplan_core.core.tabular_sources import (
CsvValueMode,
TabularColumn, TabularColumn,
TabularPreviewDiagnostic, TabularPreviewDiagnostic,
TabularPushdown, TabularPushdown,
@@ -26,6 +27,12 @@ from govoplan_core.core.tabular_sources import (
TabularSourceUnavailableError, TabularSourceUnavailableError,
TabularSourceValidationError, TabularSourceValidationError,
parse_tabular_csv, 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.core.runtime import get_registry
from govoplan_core.db.base import utcnow from govoplan_core.db.base import utcnow
@@ -281,6 +288,7 @@ class SqlTabularSourceProvider:
description: str | None = None, description: str | None = None,
delimiter: str = ",", delimiter: str = ",",
sheet_name: str | None = None, sheet_name: str | None = None,
csv_value_mode: CsvValueMode = "legacy_typed",
) -> TabularSource: ) -> TabularSource:
db, api_principal = _context(session, principal, WRITE_SCOPE) db, api_principal = _context(session, principal, WRITE_SCOPE)
inspection = self._file_adapter().inspect( inspection = self._file_adapter().inspect(
@@ -290,6 +298,7 @@ class SqlTabularSourceProvider:
file_version_id=file_version_id, file_version_id=file_version_id,
delimiter=delimiter, delimiter=delimiter,
sheet_name=sheet_name, sheet_name=sheet_name,
csv_value_mode=csv_value_mode,
) )
return self._create_origin( return self._create_origin(
db, db,
@@ -353,6 +362,7 @@ class SqlTabularSourceProvider:
file_version_id=None, file_version_id=None,
delimiter=str(item.metadata_.get("delimiter") or ","), delimiter=str(item.metadata_.get("delimiter") or ","),
sheet_name=_clean_optional(item.metadata_.get("sheet_name")), 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": elif item.provider == "postgresql":
inspection = self._sql_adapter.inspect( inspection = self._sql_adapter.inspect(
@@ -456,6 +466,20 @@ class SqlTabularSourceProvider:
f"Snapshots are limited to {MAX_SNAPSHOT_ROWS:,} rows." f"Snapshots are limited to {MAX_SNAPSHOT_ROWS:,} rows."
) )
rows = [_json_row(row) for row in snapshot.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") encoded = json.dumps(rows, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
if len(encoded) > MAX_SNAPSHOT_BYTES: if len(encoded) > MAX_SNAPSHOT_BYTES:
raise TabularSourceValidationError( raise TabularSourceValidationError(
@@ -487,7 +511,11 @@ class SqlTabularSourceProvider:
fingerprint=fingerprint, fingerprint=fingerprint,
row_count=len(rows), row_count=len(rows),
byte_count=len(encoded), 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, created_by=actor_id,
updated_by=actor_id, updated_by=actor_id,
) )
@@ -497,6 +525,15 @@ class SqlTabularSourceProvider:
db.flush() db.flush()
return _source_dto(item) 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( def delete_snapshot(
self, self,
session: object, session: object,
@@ -519,35 +556,18 @@ class SqlTabularSourceProvider:
return _source_dto(item) 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( return parse_tabular_csv(
csv_text, csv_text,
delimiter=delimiter, delimiter=delimiter,
max_rows=MAX_SNAPSHOT_ROWS, max_rows=MAX_SNAPSHOT_ROWS,
max_bytes=MAX_SNAPSHOT_BYTES,
value_mode=value_mode,
) )
def infer_schema(rows: Sequence[Mapping[str, object]]) -> tuple[TabularColumn, ...]: def infer_schema(rows: Sequence[Mapping[str, object]]) -> tuple[TabularColumn, ...]:
names: list[str] = [] return infer_tabular_schema(rows, type_name=_type_name)
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)
def snapshot_fingerprint( def snapshot_fingerprint(
@@ -717,22 +737,6 @@ def _unsupported_json(value: object) -> object:
raise TypeError(f"{type(value).__name__} is not JSON serializable") 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]: def _column_payload(column: TabularColumn) -> dict[str, object]:
return { return {
"name": column.name, "name": column.name,
+32 -3
View File
@@ -2,16 +2,45 @@ from __future__ import annotations
import tempfile import tempfile
import unittest import unittest
from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
from alembic import command
from alembic.runtime.migration import MigrationContext 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_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): 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: def test_baseline_creates_connector_tables_and_head(self) -> None:
with tempfile.TemporaryDirectory(prefix="govoplan-connectors-migration-") as directory: with tempfile.TemporaryDirectory(prefix="govoplan-connectors-migration-") as directory:
url = f"sqlite:///{Path(directory) / 'connectors.db'}" url = f"sqlite:///{Path(directory) / 'connectors.db'}"
@@ -24,7 +53,7 @@ class ConnectorsMigrationTests(unittest.TestCase):
try: try:
with engine.connect() as connection: with engine.connect() as connection:
self.assertIn( self.assertIn(
"c0f1a2b3c4d5", "d2a4c6e8f0b1",
set(MigrationContext.configure(connection).get_current_heads()), set(MigrationContext.configure(connection).get_current_heads()),
) )
self.assertTrue( self.assertTrue(
+52
View File
@@ -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())))
)
+54 -2
View File
@@ -1,23 +1,27 @@
from __future__ import annotations from __future__ import annotations
import unittest import unittest
from unittest.mock import patch
from dataclasses import replace
from fastapi import HTTPException from fastapi import HTTPException
from sqlalchemy import create_engine from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import sessionmaker
from govoplan_core.auth import ApiPrincipal from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import PrincipalRef from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.tabular_sources import ( from govoplan_core.core.tabular_sources import (
TabularCsvSource,
TabularReadRequest, TabularReadRequest,
TabularSnapshotInput, TabularSnapshotInput,
TabularSourceAccessError, TabularSourceAccessError,
TabularSourceUnavailableError, TabularSourceUnavailableError,
TabularSourceValidationError, TabularSourceValidationError,
TabularSourceNotFoundError,
) )
from govoplan_core.db.base import Base from govoplan_core.db.base import Base
from govoplan_connectors.backend.db.models import ConnectorTabularSource 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.schemas import SnapshotCreateRequest
from govoplan_connectors.backend.tabular_sources import ( from govoplan_connectors.backend.tabular_sources import (
READ_SCOPE, READ_SCOPE,
@@ -263,6 +267,54 @@ class ConnectorsTabularSourceTests(unittest.TestCase):
self.assertEqual(422, raised.exception.status_code) 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__": if __name__ == "__main__":
unittest.main() unittest.main()
+1 -1
View File
@@ -1,6 +1,6 @@
{ {
"name": "@govoplan/connectors-webui", "name": "@govoplan/connectors-webui",
"version": "0.1.26", "version": "0.1.27",
"private": true, "private": true,
"type": "module", "type": "module",
"main": "src/index.ts", "main": "src/index.ts",