fix(amazon-sns-sqs-mcp-server): add tag validation to permission mutator tools

Three permission mutator tools (add_sns_permission, add_sqs_permission,
remove_sqs_permission) bypassed the mcp_server_version tag validation
that guards all other mutative operations. This allows untagged resources
to be modified without the safety check.

Add 'validator': is_mutative_action_allowed to the tool configuration
for all three tools, consistent with other mutative operations.

Also adds @BenAtAmazon to CODEOWNERS for amazon-sns-sqs-mcp-server.

By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of the
project license.

Author: hashimsharkh

.github/CODEOWNERS

@@ -26,7 +26,7 @@ NOTICE                                                         @awslabs/mcp-admi
 /src/amazon-qbusiness-anonymous-mcp-server                     @awslabs/mcp-admins @awslabs/mcp-maintainers  @abhjaw
 /src/amazon-qindex-mcp-server                                  @awslabs/mcp-admins @awslabs/mcp-maintainers  @tkoba-aws @akhileshamara
 /src/amazon-translate-mcp-server                               @awslabs/mcp-admins @awslabs/mcp-maintainers  @goviha01 @rahullks
-/src/amazon-sns-sqs-mcp-server                                 @awslabs/mcp-admins @awslabs/mcp-maintainers  @kenliao94 @hashimsharkh
+/src/amazon-sns-sqs-mcp-server                                 @awslabs/mcp-admins @awslabs/mcp-maintainers  @kenliao94 @hashimsharkh @BenAtAmazon
 /src/aurora-dsql-mcp-server                                    @awslabs/mcp-admins @awslabs/mcp-maintainers  @anwesham-lab @benjscho @pkale @amaksimo @praba2210 @spencercorwin
 /src/aws-api-mcp-server                                        @awslabs/mcp-admins @awslabs/mcp-maintainers  @awslabs/aws-api-mcp @rshevchuk-git @PCManticore @iddv @arnewouters @bidesh
 /src/aws-appsync-mcp-server                                    @awslabs/mcp-admins @awslabs/mcp-maintainers  @phani-srikar @maxi114 @neelmurt

src/amazon-sns-sqs-mcp-server/awslabs/amazon_sns_sqs_mcp_server/sns.py

@@ -124,7 +124,10 @@ def register_sns_tools(mcp: FastMCP, disallow_resource_creation: bool = False):
 
     # Create the tool configuration dictionary
     tool_configuration = {
-        'add_permission': {'name_override': 'add_sns_permission'},
+        'add_permission': {
+            'name_override': 'add_sns_permission',
+            'validator': is_mutative_action_allowed,
+        },
         'remove_permission': {'name_override': 'remove_sns_permission'},
         'create_topic': {'func_override': create_topic_override},
         'delete_topic': {'validator': is_mutative_action_allowed},

src/amazon-sns-sqs-mcp-server/awslabs/amazon_sns_sqs_mcp_server/sqs.py

@@ -91,8 +91,14 @@ def register_sqs_tools(mcp: FastMCP, disallow_resource_creation: bool = False):
 
     # Create the tool configuration dictionary
     tool_configuration = {
-        'add_permission': {'name_override': 'add_sqs_permission'},
-        'remove_permission': {'name_override': 'remove_sqs_permission'},
+        'add_permission': {
+            'name_override': 'add_sqs_permission',
+            'validator': is_mutative_action_allowed,
+        },
+        'remove_permission': {
+            'name_override': 'remove_sqs_permission',
+            'validator': is_mutative_action_allowed,
+        },
         'create_queue': {'func_override': create_queue_override},
         'delete_queue': {'validator': is_mutative_action_allowed},
         'set_queue_attributes': {'validator': is_mutative_action_allowed},

src/amazon-sns-sqs-mcp-server/tests/test_sns.py

@@ -314,3 +314,36 @@ def test_unsubscribe_validator(self):
         )
         assert result is False
         assert message == 'mutating a resource without the mcp_server_version tag is not allowed'
+
+    @patch('boto3.client')
+    @patch('awslabs.amazon_sns_sqs_mcp_server.sns.AWSToolGenerator')
+    def test_add_permission_has_validator(self, mock_aws_tool_generator, mock_boto3_client):
+        """Test that add_permission tool has tag validation to prevent mutations on untagged resources.
+
+        This is a security-critical test: without the validator, add_sns_permission can
+        grant cross-account access to any SNS topic regardless of whether it was created
+        by the MCP server (i.e., tagged with mcp_server_version).
+        """
+        mock_mcp = MagicMock()
+        tool_config_capture = {}
+
+        def mock_generator(
+            service_name,
+            service_display_name,
+            mcp,
+            tool_configuration,
+            skip_param_documentation,
+            mcp_server_version=MCP_SERVER_VERSION,
+        ):
+            nonlocal tool_config_capture
+            tool_config_capture = tool_configuration
+            return MagicMock()
+
+        mock_aws_tool_generator.side_effect = mock_generator
+        register_sns_tools(mock_mcp)
+
+        # Verify add_permission has a validator
+        assert 'add_permission' in tool_config_capture
+        assert 'validator' in tool_config_capture['add_permission'], \
+            'add_permission must have a validator to prevent mutations on untagged resources'
+        assert tool_config_capture['add_permission']['validator'] == is_mutative_action_allowed

src/amazon-sns-sqs-mcp-server/tests/test_sqs.py

@@ -307,3 +307,69 @@ def test_is_mutative_action_allowed_exception(self):
         )
         assert result is False
         assert message == 'Test exception'
+
+    @patch('boto3.client')
+    @patch('awslabs.amazon_sns_sqs_mcp_server.sqs.AWSToolGenerator')
+    def test_add_permission_has_validator(self, mock_aws_tool_generator, mock_boto3_client):
+        """Test that add_permission tool has tag validation to prevent mutations on untagged resources.
+
+        This is a security-critical test: without the validator, add_sqs_permission can
+        grant cross-account access to any SQS queue regardless of whether it was created
+        by the MCP server (i.e., tagged with mcp_server_version).
+        """
+        mock_mcp = MagicMock()
+        tool_config_capture = {}
+
+        def mock_generator(
+            service_name,
+            service_display_name,
+            mcp,
+            tool_configuration,
+            skip_param_documentation,
+            mcp_server_version=MCP_SERVER_VERSION,
+        ):
+            nonlocal tool_config_capture
+            tool_config_capture = tool_configuration
+            return MagicMock()
+
+        mock_aws_tool_generator.side_effect = mock_generator
+        register_sqs_tools(mock_mcp)
+
+        # Verify add_permission has a validator
+        assert 'add_permission' in tool_config_capture
+        assert 'validator' in tool_config_capture['add_permission'], \
+            'add_permission must have a validator to prevent mutations on untagged resources'
+        assert tool_config_capture['add_permission']['validator'] == is_mutative_action_allowed
+
+    @patch('boto3.client')
+    @patch('awslabs.amazon_sns_sqs_mcp_server.sqs.AWSToolGenerator')
+    def test_remove_permission_has_validator(self, mock_aws_tool_generator, mock_boto3_client):
+        """Test that remove_permission tool has tag validation to prevent mutations on untagged resources.
+
+        This is a security-critical test: without the validator, remove_sqs_permission can
+        remove permissions from any SQS queue regardless of whether it was created
+        by the MCP server (i.e., tagged with mcp_server_version).
+        """
+        mock_mcp = MagicMock()
+        tool_config_capture = {}
+
+        def mock_generator(
+            service_name,
+            service_display_name,
+            mcp,
+            tool_configuration,
+            skip_param_documentation,
+            mcp_server_version=MCP_SERVER_VERSION,
+        ):
+            nonlocal tool_config_capture
+            tool_config_capture = tool_configuration
+            return MagicMock()
+
+        mock_aws_tool_generator.side_effect = mock_generator
+        register_sqs_tools(mock_mcp)
+
+        # Verify remove_permission has a validator
+        assert 'remove_permission' in tool_config_capture
+        assert 'validator' in tool_config_capture['remove_permission'], \
+            'remove_permission must have a validator to prevent mutations on untagged resources'
+        assert tool_config_capture['remove_permission']['validator'] == is_mutative_action_allowed