diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a57a4c1ca..a42b7184c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,9 @@ ENHANCEMENTS: +BUG FIXES: +* Fix to enhance service bus handling of invalid JSON in receive_message function ([#4932](https://github.com/microsoft/AzureTRE/pull/4932)) + ## (0.29.0) (August 14, 2026) **BREAKING CHANGES** * Remove Windows 10 and dsvm image support from Guacamole. ([#4890](https://github.com/microsoft/AzureTRE/issues/4890)) diff --git a/resource_processor/_version.py b/resource_processor/_version.py index e318db3960..af935ca6d9 100644 --- a/resource_processor/_version.py +++ b/resource_processor/_version.py @@ -1 +1 @@ -__version__ = "0.13.6" +__version__ = "0.13.7" diff --git a/resource_processor/tests_rp/test_runner.py b/resource_processor/tests_rp/test_runner.py index 6c9166b017..ae667de312 100644 --- a/resource_processor/tests_rp/test_runner.py +++ b/resource_processor/tests_rp/test_runner.py @@ -115,7 +115,15 @@ async def test_receive_message(mock_invoke_porter_action, mock_service_bus_clien mock_receiver.__aexit__.return_value = None mock_receiver.session.session_id = "test_session_id" mock_receiver.__aiter__.return_value = [AsyncMock()] - mock_receiver.__aiter__.return_value[0] = json.dumps({"id": "test_id", "action": "install", "stepId": "test_step_id", "operationId": "test_operation_id"}) + mock_receiver.__aiter__.return_value[0] = json.dumps({ + "id": "test_id", + "action": "install", + "stepId": "test_step_id", + "operationId": "test_operation_id", + "name": "test_bundle", + "version": "1.0.0", + "parameters": {}, + }) mock_service_bus_client_instance.get_queue_receiver.return_value.__aenter__.return_value = mock_receiver @@ -124,10 +132,130 @@ async def test_receive_message(mock_invoke_porter_action, mock_service_bus_clien config = {"resource_request_queue": "test_queue"} await receive_message(mock_service_bus_client_instance, config, keep_running=run_once) - mock_receiver.complete_message.assert_called_once() + mock_receiver.complete_message.assert_awaited_once() mock_service_bus_client_instance.get_queue_receiver.assert_called_once_with(queue_name="test_queue", max_wait_time=1, session_id=ServiceBusSessionFilter.NEXT_AVAILABLE) +@pytest.mark.asyncio +async def test_receive_message_bad_json(mock_service_bus_client, mock_auto_lock_renewer): + mock_service_bus_client_instance = mock_service_bus_client.return_value + + # Set up the lock renewer mock correctly + mock_renewer = AsyncMock() + mock_renewer.register = Mock() + mock_auto_lock_renewer.return_value.__aenter__.return_value = mock_renewer + + mock_receiver = AsyncMock() + mock_receiver.__aenter__.return_value = mock_receiver + mock_receiver.__aexit__.return_value = None + mock_receiver.session.session_id = "test_session_id" + mock_receiver.__aiter__.return_value = ["invalid_json_string"] + + mock_service_bus_client_instance.get_queue_receiver.return_value.__aenter__.return_value = mock_receiver + + run_once = Mock(side_effect=[True, False]) + + config = {"resource_request_queue": "test_queue"} + + await receive_message(mock_service_bus_client_instance, config, keep_running=run_once) + mock_receiver.dead_letter_message.assert_awaited_once() + dead_letter_args, dead_letter_kwargs = mock_receiver.dead_letter_message.await_args + assert dead_letter_args == ("invalid_json_string",) + assert dead_letter_kwargs["reason"] == "InvalidJSON" + assert dead_letter_kwargs["error_description"] + mock_receiver.complete_message.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "message, expected_log", + [ + ("invalid_json_string", "Failed to dead-letter malformed message"), + ("null", "Failed to dead-letter invalid resource request message"), + ], +) +async def test_receive_message_dead_letter_failure_is_logged( + message, expected_log, mock_service_bus_client, mock_auto_lock_renewer, mock_logger +): + mock_service_bus_client_instance = mock_service_bus_client.return_value + + mock_renewer = AsyncMock() + mock_renewer.register = Mock() + mock_auto_lock_renewer.return_value.__aenter__.return_value = mock_renewer + + mock_receiver = AsyncMock() + mock_receiver.__aenter__.return_value = mock_receiver + mock_receiver.__aexit__.return_value = None + mock_receiver.session.session_id = "test_session_id" + mock_receiver.__aiter__.return_value = [message] + mock_receiver.dead_letter_message.side_effect = RuntimeError("lock lost") + + mock_service_bus_client_instance.get_queue_receiver.return_value.__aenter__.return_value = mock_receiver + + run_once = Mock(side_effect=[True, False]) + config = {"resource_request_queue": "test_queue"} + + await receive_message(mock_service_bus_client_instance, config, keep_running=run_once) + + mock_logger.exception.assert_called_once_with(expected_log) + mock_receiver.complete_message.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "message, error_description", + [ + ("null", "Resource request message must be a JSON object"), + ("[]", "Resource request message must be a JSON object"), + ("\"text\"", "Resource request message must be a JSON object"), + ( + '{"id": "test_id"}', + "Resource request message is missing fields: ['action', 'name', 'operationId', 'parameters', 'stepId', 'version']", + ), + ( + '{"id": "test_id", "action": [], "stepId": "test_step_id", "operationId": "test_operation_id", "name": "test_bundle", "version": "1.0.0", "parameters": {}}', + "Resource request message has invalid field types: ['action']", + ), + ( + '{"id": "test_id", "action": "install", "stepId": "test_step_id", "operationId": "test_operation_id", "name": "test_bundle", "version": "1.0.0", "parameters": null}', + "Resource request message has invalid field types: ['parameters']", + ), + ( + '{"id": "test_id", "action": "install", "stepId": "test_step_id", "operationId": "test_operation_id", "name": "test_bundle", "version": "1.0.0", "parameters": {}, "user": []}', + "Resource request message has invalid field types: ['user']", + ), + ], +) +async def test_receive_message_invalid_json_structure( + message, error_description, mock_service_bus_client, mock_auto_lock_renewer +): + mock_service_bus_client_instance = mock_service_bus_client.return_value + + mock_renewer = AsyncMock() + mock_renewer.register = Mock() + mock_auto_lock_renewer.return_value.__aenter__.return_value = mock_renewer + + mock_receiver = AsyncMock() + mock_receiver.__aenter__.return_value = mock_receiver + mock_receiver.__aexit__.return_value = None + mock_receiver.session.session_id = "test_session_id" + mock_receiver.__aiter__.return_value = [message] + + mock_service_bus_client_instance.get_queue_receiver.return_value.__aenter__.return_value = mock_receiver + + run_once = Mock(side_effect=[True, False]) + config = {"resource_request_queue": "test_queue"} + + await receive_message(mock_service_bus_client_instance, config, keep_running=run_once) + + mock_receiver.dead_letter_message.assert_awaited_once_with( + message, + reason="InvalidResourceRequest", + error_description=error_description, + ) + mock_receiver.complete_message.assert_not_awaited() + + @pytest.mark.asyncio async def test_receive_message_unknown_exception(mock_auto_lock_renewer, mock_service_bus_client, mock_logger): """Test receiving a message with an unknown exception.""" @@ -143,7 +271,15 @@ async def test_receive_message_unknown_exception(mock_auto_lock_renewer, mock_se mock_receiver.__aexit__.return_value = None mock_receiver.session.session_id = "test_session_id" mock_receiver.__aiter__.return_value = [AsyncMock()] - mock_receiver.__aiter__.return_value[0] = json.dumps({"id": "test_id", "action": "install", "stepId": "test_step_id", "operationId": "test_operation_id"}) + mock_receiver.__aiter__.return_value[0] = json.dumps({ + "id": "test_id", + "action": "install", + "stepId": "test_step_id", + "operationId": "test_operation_id", + "name": "test_bundle", + "version": "1.0.0", + "parameters": {}, + }) mock_service_bus_client_instance.get_queue_receiver.return_value.__aenter__.return_value = mock_receiver diff --git a/resource_processor/vmss_porter/runner.py b/resource_processor/vmss_porter/runner.py index 120ececba0..2b1eb45eef 100644 --- a/resource_processor/vmss_porter/runner.py +++ b/resource_processor/vmss_porter/runner.py @@ -18,6 +18,43 @@ from azure.identity.aio import DefaultAzureCredential +RESOURCE_REQUEST_FIELD_TYPES = { + "id": str, + "action": str, + "stepId": str, + "operationId": str, + "name": str, + "version": str, + "parameters": dict, +} +OPTIONAL_RESOURCE_REQUEST_FIELD_TYPES = {"user": dict} + + +def validate_resource_request(message: object) -> None: + if not isinstance(message, dict): + raise ValueError("Resource request message must be a JSON object") + + missing_fields = set(RESOURCE_REQUEST_FIELD_TYPES) - message.keys() + if missing_fields: + raise ValueError(f"Resource request message is missing fields: {sorted(missing_fields)}") + + invalid_fields = [ + field_name + for field_name, field_type in RESOURCE_REQUEST_FIELD_TYPES.items() + if not isinstance(message[field_name], field_type) + ] + if invalid_fields: + raise ValueError(f"Resource request message has invalid field types: {sorted(invalid_fields)}") + + invalid_optional_fields = [ + field_name + for field_name, field_type in OPTIONAL_RESOURCE_REQUEST_FIELD_TYPES.items() + if field_name in message and not isinstance(message[field_name], field_type) + ] + if invalid_optional_fields: + raise ValueError(f"Resource request message has invalid field types: {sorted(invalid_optional_fields)}") + + def set_up_config() -> Optional[dict]: try: config = get_config() @@ -71,8 +108,23 @@ async def receive_message(service_bus_client, config: dict, keep_running=lambda: try: message = json.loads(str(msg)) - except (json.JSONDecodeError) as e: + except json.JSONDecodeError as e: logger.error(f"Received bad service bus resource request message: {e}") + try: + await receiver.dead_letter_message(msg, reason="InvalidJSON", error_description=str(e)) + except Exception: + logger.exception("Failed to dead-letter malformed message") + continue + + try: + validate_resource_request(message) + except ValueError as e: + logger.error(f"Received invalid service bus resource request message: {e}") + try: + await receiver.dead_letter_message(msg, reason="InvalidResourceRequest", error_description=str(e)) + except Exception: + logger.exception("Failed to dead-letter invalid resource request message") + continue with tracer.start_as_current_span("receive_message") as current_span: current_span.set_attribute("resource_id", message["id"])