diff --git a/tests/integration/server_rest_api/datex2/datex2_import_api_test.py b/tests/integration/server_rest_api/datex2/datex2_import_api_test.py index 78db676..56cfabd 100644 --- a/tests/integration/server_rest_api/datex2/datex2_import_api_test.py +++ b/tests/integration/server_rest_api/datex2/datex2_import_api_test.py @@ -337,6 +337,48 @@ def test_push_realtime_v35_large_delta_push_is_async( evse_charging = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first() assert evse_charging.status == EvseStatus.UNKNOWN + @staticmethod + def test_push_realtime_v35_oversized_payload_skips_validation_and_queues( + db: SQLAlchemy, + test_client: OpenApiFlaskClient, + requests_mock: Mocker, + isolated_datex2_dir: Path, + stubbed_celery_delay, + ) -> None: + """ + Payloads larger than DATEX2_SYNC_MAX_CONTENT_LENGTH must be handed off to celery without + being parsed or validated on the request thread - so even a structurally invalid payload is + persisted + queued (HTTP 200) instead of being rejected with HTTP 400. + """ + _import_static_data(requests_mock) + + # A payload that would normally be rejected as invalid, but is pushed over the size limit. + invalid_payload = {'not': 'a valid messageContainer'} + + config = dependencies.get_config_helper().get_config() + sentinel = object() + original = config.get('DATEX2_SYNC_MAX_CONTENT_LENGTH', sentinel) + config['DATEX2_SYNC_MAX_CONTENT_LENGTH'] = 5 + try: + response = test_client.post( + path=f'/api/server/v1/datex/v3.5/{SOURCE_UID}/realtime?key={API_KEY}', + json=invalid_payload, + ) + finally: + if original is sentinel: + config.pop('DATEX2_SYNC_MAX_CONTENT_LENGTH', None) + else: + config['DATEX2_SYNC_MAX_CONTENT_LENGTH'] = original + + assert response.status_code == HTTPStatus.OK + + # Handed off to celery unvalidated: payload persisted to disk and task queued. + written = list(isolated_datex2_dir.iterdir()) + assert len(written) == 1 + with written[0].open('rb') as f: + assert json.load(f) == invalid_payload + stubbed_celery_delay.assert_called_once() + @staticmethod def test_push_realtime_v35_invalid_payload_returns_400( db: SQLAlchemy, diff --git a/webapp/common/constants.py b/webapp/common/constants.py index 84ec54f..3b97230 100644 --- a/webapp/common/constants.py +++ b/webapp/common/constants.py @@ -42,6 +42,10 @@ class BaseConfig: # processed asynchronously via celery instead of inline on the request thread (snapshots are always async). DATEX2_ASYNC_STATION_STATUS_THRESHOLD = 25 + # DATEX II realtime pushes with a request body larger than this many bytes are handed off to celery + # without being parsed or validated on the request thread, so large payloads never block it. + DATEX2_SYNC_MAX_CONTENT_LENGTH = 250 * 1024 + PUBLIC_IMAGE_PATH = '/data/images' SQLALCHEMY_TRACK_MODIFICATIONS = False diff --git a/webapp/server_rest_api/datex2/datex2_rest_api.py b/webapp/server_rest_api/datex2/datex2_rest_api.py index 3fc429f..5a49ab9 100644 --- a/webapp/server_rest_api/datex2/datex2_rest_api.py +++ b/webapp/server_rest_api/datex2/datex2_rest_api.py @@ -104,12 +104,14 @@ def head(self, source_uid: str) -> tuple[Response, HTTPStatus]: summary='Push DATEX II v3.5 realtime EVSE statuses', description=( 'Accepts a Mobilithek-style DATEX II v3.5 JSON payload (`messageContainer.payload[0]` ' - '`aegiEnergyInfrastructureStatusPublication`). The payload is validated synchronously; ' - 'small `deltaPush` updates are applied directly, while snapshots (any other exchange ' - 'protocol) and `deltaPush` payloads with more than `DATEX2_ASYNC_STATION_STATUS_THRESHOLD` ' - '`energyInfrastructureStationStatus` entries are persisted to `DATEX2_IMPORT_DIR` and ' - 'processed asynchronously by a Celery worker so very large payloads do not block the ' - "request thread. The `key` query parameter must match the source's configured `api_key`." + '`aegiEnergyInfrastructureStatusPublication`). Payloads larger than ' + '`DATEX2_SYNC_MAX_CONTENT_LENGTH` bytes are persisted to `DATEX2_IMPORT_DIR` and processed ' + 'asynchronously by a Celery worker without being parsed or validated on the request thread. ' + 'Smaller payloads are validated synchronously; small `deltaPush` updates are applied ' + 'directly, while snapshots (any other exchange protocol) and `deltaPush` payloads with more ' + 'than `DATEX2_ASYNC_STATION_STATUS_THRESHOLD` `energyInfrastructureStationStatus` entries are ' + 'persisted and processed asynchronously as well. The `key` query parameter must match the ' + "source's configured `api_key`." ), path=[Parameter('source_uid', schema=StringField())], query=[ @@ -138,6 +140,15 @@ def post(self, source_uid: str) -> tuple[Response, HTTPStatus]: if not data: raise InputValidationException(message='no realtime payload') + # Parsing and validating a large payload synchronously is expensive and would block the + # request thread, so anything above DATEX2_SYNC_MAX_CONTENT_LENGTH bytes is handed off to a + # celery worker unparsed and unvalidated - the worker parses, validates and applies it. + sync_max_content_length = self.config_helper.get('DATEX2_SYNC_MAX_CONTENT_LENGTH', 250 * 1024) + if len(data) > sync_max_content_length: + self._queue_for_async_processing(source_uid, data) + # Mobilithek expects HTTP 200 instead of HTTP 204 + return empty_json_response(), HTTPStatus.OK + try: parsed_data = json.loads(data) except json.JSONDecodeError as e: @@ -158,6 +169,16 @@ def post(self, source_uid: str) -> tuple[Response, HTTPStatus]: # Mobilithek expects HTTP 200 instead of HTTP 204 return empty_json_response(), HTTPStatus.OK + self._queue_for_async_processing(source_uid, data) + + # Mobilithek expects HTTP 200 instead of HTTP 204 + return empty_json_response(), HTTPStatus.OK + + def _queue_for_async_processing(self, source_uid: str, data: bytes) -> None: + """ + Persist a raw realtime payload to ``DATEX2_IMPORT_DIR`` and hand it off to a celery worker + for asynchronous parsing, validation and application via ``datex2_v3_5_realtime_import_by_file``. + """ base_path = Path(self.config_helper.get('DATEX2_IMPORT_DIR')) if not base_path.is_dir(): base_path.mkdir(parents=True, exist_ok=True) @@ -168,9 +189,6 @@ def post(self, source_uid: str) -> tuple[Response, HTTPStatus]: self.celery_helper.delay(datex2_v3_5_realtime_import_by_file, source_uid, str(import_path)) - # Mobilithek expects HTTP 200 instead of HTTP 204 - return empty_json_response(), HTTPStatus.OK - def _dump_request(self, source_uid: str, request_body: bytes) -> None: source_config = self.config_helper.get('SOURCES', {}).get(source_uid) or {} if source_config.get('debug', False) is False: