diff --git a/docs/CONNECTOR_SOURCE_LIFECYCLE.md b/docs/CONNECTOR_SOURCE_LIFECYCLE.md index 2171b8c..a818532 100644 --- a/docs/CONNECTOR_SOURCE_LIFECYCLE.md +++ b/docs/CONNECTOR_SOURCE_LIFECYCLE.md @@ -132,6 +132,25 @@ explicit refresh. Missing Files capability, revoked file access, quarantine, oversized or malformed content, inactive/stale credentials, unreachable SQL, and timeout failures produce sanitized unavailable/validation diagnostics. +XLSX archive checks and workbook parsing run in a fresh disposable Core worker. +Each invocation allows 15 seconds wall time, 10 CPU seconds and 512 MiB virtual +address space, with no file output. The typed transport allows 8 MiB input and +64 MiB result bytes, at most 64 nesting levels and 1,000,000 value nodes. These +transport bounds include serialization overhead. The existing workbook bounds +remain 5,000,000 raw bytes, 50,000,000 expanded bytes, 5,000 archive entries, +100:1 compression ratio, 500 columns and 10,000 row positions after the header. +Authorization and exact-version file reads happen in the parent; credentials, +SQL sessions and durable changes are never passed to the parser. + +Users receive an explicit failure rather than a partial source when resource +or transport limits are exceeded; reduce workbook size or complexity before +retrying. The Core `GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY` setting limits active work per +API/worker process without queuing. Busy capacity may be retried later. Missing +POSIX resource controls, cancellation and worker failure produce sanitized +unavailable diagnostics; there is no in-process fallback. Operators must keep +the Core worker API available and account for the aggregate memory of all +active worker slots across API/worker replicas. + All three current providers declare projection and pagination pushdown only. Filters, aggregations, and sorting remain in Dataflow until an adapter explicitly declares and tests those operations. diff --git a/src/govoplan_connectors/backend/manifest.py b/src/govoplan_connectors/backend/manifest.py index 5e8e84a..0924d10 100644 --- a/src/govoplan_connectors/backend/manifest.py +++ b/src/govoplan_connectors/backend/manifest.py @@ -1130,6 +1130,44 @@ manifest = ModuleManifest( related_modules=("dataflow", "files", "reporting", "risk_compliance"), order=40, ), + DocumentationTopic( + id="connectors.xlsx-worker-limits", + title="XLSX processing resource limits", + summary="Understand isolated workbook parsing and explicit capacity failures.", + body=( + "Authorized XLSX bytes are parsed in a fresh disposable process with a 15-second wall limit, 10 CPU seconds, " + "512 MiB address-space limit, no file output, and 8 MiB input/64 MiB result transport ceilings. Existing limits remain " + "5,000,000 input bytes, 50,000,000 expanded archive bytes, 5,000 archive entries, 100:1 expansion ratio, " + "500 columns and 10,000 row positions after the header. Typed transport additionally limits nesting to 64 levels " + "and 1,000,000 value nodes. Limit failures reject the whole parse; reduce the workbook before retrying. " + "The shared Core GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY capacity is per API/worker process and does not queue: busy work " + "fails explicitly and may be retried later. POSIX resource controls are required; missing controls, cancellation " + "or worker failure produce sanitized unavailable diagnostics. No in-process fallback occurs. File authorization, " + "credentials, SQL sessions and durable source changes remain in the parent. Operators must budget aggregate " + "memory across process slots and keep the required Core worker API available." + ), + layer="static", + documentation_types=("user", "admin"), + audience=("user", "module_admin", "operator"), + translations={"de": { + "title": "Ressourcengrenzen der XLSX-Verarbeitung", + "summary": "Isoliertes Einlesen von Arbeitsmappen und ausdrückliche Kapazitätsfehler verstehen.", + "body": ( + "Autorisierte XLSX-Bytes werden in einem neuen kurzlebigen Prozess verarbeitet: höchstens 15 Sekunden Gesamtdauer, " + "10 CPU-Sekunden, 512 MiB Adressraum, keine Dateiausgabe und 8 MiB Eingabe-/64 MiB Ergebnistransport. Weiterhin gelten " + "5.000.000 Eingabebytes, 50.000.000 entpackte Archivbytes, 5.000 Archiveinträge, ein Entpackverhältnis von 100:1, " + "500 Spalten und 10.000 Zeilenpositionen nach der Kopfzeile. Der typisierte Transport begrenzt zusätzlich die " + "Verschachtelung auf 64 Ebenen und 1.000.000 Wertknoten. Grenzverletzungen lehnen den gesamten Lesevorgang ab; " + "verkleinern Sie die Arbeitsmappe vor einem erneuten Versuch. Die gemeinsame Core-Einstellung GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY " + "gilt pro API-/Worker-Prozess und bildet keine Warteschlange: Bei belegter Kapazität erfolgt ein ausdrücklicher Fehler; " + "versuchen Sie es später erneut. POSIX-Ressourcenbegrenzungen sind erforderlich. Fehlende Kontrollen, Abbruch oder " + "Worker-Fehler melden eine bereinigte Nichtverfügbarkeit. Es gibt keinen Rückfall auf Verarbeitung im Elternprozess. " + "Dateiberechtigungen, Zugangsdaten, SQL-Sitzungen und dauerhafte Quelländerungen bleiben im Elternprozess. Betreiber " + "müssen den Gesamtspeicher aller Prozessplätze berücksichtigen und die benötigte Core-Worker-API bereitstellen." + ), + }}, + order=41, + ), DocumentationTopic( id="connectors.rss-atom", title="RSS and Atom feeds", diff --git a/src/govoplan_connectors/backend/tabular_adapters.py b/src/govoplan_connectors/backend/tabular_adapters.py index f39dfc8..3bf0fe7 100644 --- a/src/govoplan_connectors/backend/tabular_adapters.py +++ b/src/govoplan_connectors/backend/tabular_adapters.py @@ -59,7 +59,17 @@ from govoplan_core.security.credential_envelopes import ( CredentialEnvelopeError, resolve_credential_envelope, ) +from govoplan_core.security.bounded_process import ( + ProcessBudgetError, + ProcessLimits, + run_bounded_operation, +) from govoplan_core.security.redaction import is_sensitive_key +from govoplan_core.security.worker_payload import ( + WorkerPayloadError, + decode_worker_payload, + encode_worker_payload, +) from govoplan_connectors.backend.db.models import ConnectorConfiguration @@ -69,6 +79,13 @@ MAX_FILE_COLUMNS = 500 MAX_XLSX_ENTRIES = 5_000 MAX_XLSX_EXPANDED_BYTES = 50_000_000 MAX_XLSX_COMPRESSION_RATIO = 100 +XLSX_PROCESS_LIMITS = ProcessLimits( + wall_seconds=15, + cpu_seconds=10, + memory_bytes=512 * 1024 * 1024, + input_bytes=8 * 1024 * 1024, + output_bytes=64 * 1024 * 1024, +) POSTGRESQL_SCHEMES = frozenset({"postgresql", "postgresql+psycopg"}) _IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_$-]{0,127}$") @@ -573,6 +590,55 @@ def _parse_xlsx( payload: bytes, *, sheet_name: str | None, +) -> tuple[tuple[Mapping[str, object], ...], str]: + # Only already-authorized workbook bytes and a worksheet selector leave + # the parent; Files access, credentials and SQL sessions never cross. + try: + encoded = encode_worker_payload( + {"payload": payload, "sheet_name": sheet_name}, + max_bytes=XLSX_PROCESS_LIMITS.input_bytes, + ) + result = decode_worker_payload( + run_bounded_operation(_parse_xlsx_worker, encoded, limits=XLSX_PROCESS_LIMITS), + max_bytes=XLSX_PROCESS_LIMITS.output_bytes, + ) + except ProcessBudgetError as exc: + error_type = ( + TabularSourceUnavailableError + if exc.code in {"busy", "cancelled", "unavailable", "worker_failed"} + else TabularSourceValidationError + ) + raise error_type(f"Managed XLSX processing failed ({exc.code}): {exc}") from exc + except (TypeError, ValueError, RecursionError) as exc: + raise TabularSourceValidationError("Managed XLSX data could not be safely transferred.") from exc + if not isinstance(result, dict): + raise TabularSourceUnavailableError("Managed XLSX worker returned an invalid result.") + if "validation_error" in result: + raise TabularSourceValidationError(str(result["validation_error"])) + rows, selected_name = result.get("rows"), result.get("sheet_name") + if not isinstance(rows, tuple) or not isinstance(selected_name, str): + raise TabularSourceUnavailableError("Managed XLSX worker returned an invalid result.") + return rows, selected_name + + +def _parse_xlsx_worker(payload: bytes) -> bytes: + data = decode_worker_payload(payload, max_bytes=XLSX_PROCESS_LIMITS.input_bytes) + try: + rows, sheet_name = _parse_xlsx_content(data["payload"], sheet_name=data["sheet_name"]) + return encode_worker_payload( + {"rows": rows, "sheet_name": sheet_name}, + max_bytes=XLSX_PROCESS_LIMITS.output_bytes, + ) + except TabularSourceValidationError as exc: + return encode_worker_payload({"validation_error": str(exc)}) + except WorkerPayloadError: + return encode_worker_payload({"validation_error": "Managed XLSX parsed data exceeds the worker transport limit."}) + + +def _parse_xlsx_content( + payload: bytes, + *, + sheet_name: str | None, ) -> tuple[tuple[Mapping[str, object], ...], str]: _validate_xlsx_archive(payload) try: @@ -582,6 +648,8 @@ def _parse_xlsx( data_only=True, keep_links=False, ) + except MemoryError: + raise except Exception as exc: raise TabularSourceValidationError( "Managed XLSX content could not be parsed." diff --git a/tests/test_xlsx_safety_bounds.py b/tests/test_xlsx_safety_bounds.py index b6a386d..85d4304 100644 --- a/tests/test_xlsx_safety_bounds.py +++ b/tests/test_xlsx_safety_bounds.py @@ -2,6 +2,7 @@ from __future__ import annotations import re import unittest +from dataclasses import replace from io import BytesIO from unittest.mock import patch from zipfile import ZipFile @@ -40,14 +41,12 @@ def parse(payload: bytes): class XlsxSafetyBoundsTests(unittest.TestCase): def test_sparse_rows_cannot_bypass_limit_by_not_counting_as_data(self): - with patch.object(adapters, "MAX_FILE_ROWS", 3): - with self.assertRaisesRegex(TabularSourceValidationError, "row"): - parse(workbook_bytes(last_cell="A6")) + with self.assertRaisesRegex(TabularSourceValidationError, "row"): + parse(workbook_bytes(last_cell=f"A{adapters.MAX_FILE_ROWS + 2}")) def test_forged_small_dimensions_cannot_hide_far_away_cells(self): - with patch.object(adapters, "MAX_FILE_ROWS", 3): - with self.assertRaisesRegex(TabularSourceValidationError, "row"): - parse(workbook_bytes(dimension="A1:A2", last_cell="A1000000")) + with self.assertRaisesRegex(TabularSourceValidationError, "row"): + parse(workbook_bytes(dimension="A1:A2", last_cell="A1000000")) def test_forged_small_dimensions_cannot_hide_out_of_range_columns(self): with self.assertRaisesRegex(TabularSourceValidationError, "column"): @@ -61,10 +60,31 @@ class XlsxSafetyBoundsTests(unittest.TestCase): self.assertEqual("Sheet", sheet) def test_small_blank_gaps_and_exact_limit_remain_usable(self): - with patch.object(adapters, "MAX_FILE_ROWS", 3): - rows, _sheet = parse(workbook_bytes(last_cell="A4")) + for cell in ("A4", f"A{adapters.MAX_FILE_ROWS + 1}"): + rows, _sheet = parse(workbook_bytes(last_cell=cell)) self.assertEqual(({"name": "Ada"},), rows) + def test_workbook_is_parsed_in_a_fresh_child_not_the_parent(self): + with patch.object(adapters, "_parse_xlsx_content", side_effect=AssertionError("parent parser ran")): + rows, sheet = parse(workbook_bytes()) + self.assertEqual(({"name": "Ada"},), rows) + self.assertEqual("Sheet", sheet) + + def test_real_worker_timeout_fails_without_parent_fallback(self): + limits = replace(adapters.XLSX_PROCESS_LIMITS, wall_seconds=0.001) + with patch.object(adapters, "XLSX_PROCESS_LIMITS", limits), patch.object( + adapters, "_parse_xlsx_content", side_effect=AssertionError("parent parser ran") + ): + with self.assertRaisesRegex(TabularSourceValidationError, "timeout"): + parse(workbook_bytes()) + + def test_child_validation_keeps_missing_sheet_message(self): + with self.assertRaisesRegex(TabularSourceValidationError, "worksheet 'Missing' was not found"): + adapters.parse_managed_tabular_content( + workbook_bytes(), filename="fixture.xlsx", content_type=None, + delimiter=",", sheet_name="Missing", + ) + if __name__ == "__main__": unittest.main()