add testing for api key authentication

Author: Edwardius

robot_mcp_server/CMakeLists.txt

@@ -132,6 +132,10 @@ if(BUILD_TESTING)
   add_launch_test(
     test/launch_tests/test_http_integration.py
   )
+
+  add_launch_test(
+    test/launch_tests/test_authentication.py
+  )
 endif()
 
 ament_export_targets(${PROJECT_NAME}Targets HAS_LIBRARY_TARGET)

robot_mcp_server/src/mcp_config/config_parser.cpp

@@ -91,9 +91,10 @@ ServerConfig ConfigParser::parseServerConfig(rclcpp_lifecycle::LifecycleNode::Sh
   config.bond_timeout = node->declare_parameter("server.bond_timeout", config.bond_timeout);
   config.bond_heartbeat_period = node->declare_parameter("server.bond_heartbeat_period", config.bond_heartbeat_period);
 
-  // Optional API key
-  if (node->has_parameter("server.api_key")) {
-    config.api_key = node->get_parameter("server.api_key").as_string();
+  // Optional API key - declare with empty string default
+  std::string api_key_str = node->declare_parameter("server.api_key", std::string(""));
+  if (!api_key_str.empty()) {
+    config.api_key = api_key_str;
   }
 
   return config;

robot_mcp_server/test/launch_tests/test_authentication.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+# Copyright (c) 2025-present WATonomous. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Integration tests for MCP server authentication."""
+
+import time
+import unittest
+
+import launch
+import launch.actions
+import launch_ros.actions
+import launch_testing.actions
+import pytest
+import requests
+
+
+@pytest.mark.launch_test
+def generate_test_description():
+    """Launch the MCP server with authentication enabled."""
+    # Test API key
+    test_api_key = "test_secret_key_12345"
+
+    mcp_server_node = launch_ros.actions.LifecycleNode(
+        package="robot_mcp_server",
+        executable="robot_mcp_server_node",
+        name="mcp_http_server",
+        namespace="",
+        parameters=[
+            {
+                "server.host": "127.0.0.1",
+                "server.port": 18081,
+                "server.api_key": test_api_key,
+                "server.enable_https": False,
+            }
+        ],
+        output="screen",
+    )
+
+    # Use nav2_lifecycle_manager to automatically activate the node
+    lifecycle_manager = launch_ros.actions.Node(
+        package="nav2_lifecycle_manager",
+        executable="lifecycle_manager",
+        name="test_lifecycle_manager",
+        parameters=[
+            {"autostart": True, "node_names": ["mcp_http_server"], "bond_timeout": 4.0}
+        ],
+        output="screen",
+    )
+
+    return launch.LaunchDescription(
+        [
+            mcp_server_node,
+            lifecycle_manager,
+            launch_testing.actions.ReadyToTest(),
+        ]
+    )
+
+
+class TestAuthentication(unittest.TestCase):
+    """Active tests for authentication."""
+
+    BASE_URL = "http://127.0.0.1:18081"
+    VALID_API_KEY = "test_secret_key_12345"
+
+    @classmethod
+    def setUpClass(cls):
+        """Wait for server to be ready."""
+        time.sleep(5)
+
+    def test_valid_api_key(self):
+        """Test request with valid API key succeeds."""
+        payload = {"jsonrpc": "2.0", "method": "test", "id": 1}
+        headers = {"Authorization": f"Bearer {self.VALID_API_KEY}"}
+
+        response = requests.post(f"{self.BASE_URL}/mcp", json=payload, headers=headers)
+
+        assert response.status_code == 200
+        data = response.json()
+        assert "result" in data or "message" in data
+
+    def test_invalid_api_key(self):
+        """Test request with invalid API key is rejected."""
+        payload = {"jsonrpc": "2.0", "method": "test", "id": 1}
+        headers = {"Authorization": "Bearer wrong_key"}
+
+        response = requests.post(f"{self.BASE_URL}/mcp", json=payload, headers=headers)
+
+        assert response.status_code == 401
+        data = response.json()
+        assert "error" in data
+
+    def test_missing_authorization_header(self):
+        """Test request without Authorization header is rejected."""
+        payload = {"jsonrpc": "2.0", "method": "test", "id": 1}
+
+        response = requests.post(f"{self.BASE_URL}/mcp", json=payload)
+
+        assert response.status_code == 401
+        data = response.json()
+        assert "error" in data
+
+    def test_malformed_authorization_header(self):
+        """Test request with malformed Authorization header is rejected."""
+        payload = {"jsonrpc": "2.0", "method": "test", "id": 1}
+        headers = {"Authorization": "InvalidFormat"}
+
+        response = requests.post(f"{self.BASE_URL}/mcp", json=payload, headers=headers)
+
+        assert response.status_code == 401
+        data = response.json()
+        assert "error" in data
+
+    def test_authorization_header_without_bearer(self):
+        """Test request with Authorization header but no Bearer prefix is rejected."""
+        payload = {"jsonrpc": "2.0", "method": "test", "id": 1}
+        headers = {"Authorization": self.VALID_API_KEY}
+
+        response = requests.post(f"{self.BASE_URL}/mcp", json=payload, headers=headers)
+
+        assert response.status_code == 401
+        data = response.json()
+        assert "error" in data
+
+
+@launch_testing.post_shutdown_test()
+class TestAuthenticationShutdown(unittest.TestCase):
+    """Post-shutdown tests."""
+
+    def test_exit_codes(self, proc_info):
+        """Check that all processes exited cleanly."""
+        launch_testing.asserts.assertExitCodes(proc_info)