Enhance service bus handling of invalid JSON in receive_message function

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))

resource_processor/_version.py

@@ -1 +1 @@
-__version__ = "0.13.6"
+__version__ = "0.13.7"

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
 

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"])