diff --git a/server/gti/README.md b/server/gti/README.md index 662bfd0a..f35a0bdb 100644 --- a/server/gti/README.md +++ b/server/gti/README.md @@ -6,6 +6,13 @@ Threat Intelligence suite. ## Features +### Tool Annotations & Safety Hints + +All tools exposed by the Google Threat Intelligence MCP server declare Model Context Protocol `ToolAnnotations` metadata to guide autonomous clients (such as Gemini CLI, Google ADK, Claude, and Cursor) regarding execution safety: + +- **Read-Only Tools (`readOnlyHint: true`)**: Threat reporting, entity lookups, IOC searches, domain and IP intelligence, threat profiles, and ruleset queries (32 tools total). +- **Additive / Mutating Tools (`readOnlyHint: false, destructiveHint: false`)**: Operations that create or update threat collections or request file analysis (`create_collection`, `update_collection_attributes`, `update_iocs_in_collection`, `analyse_file`). + ### Collections (Threats) - **`get_collection_report(id)`**: Retrieves a specific collection report by its ID (e.g., `report--`, `threat-actor--`). diff --git a/server/gti/gti_mcp/tools/collections.py b/server/gti/gti_mcp/tools/collections.py index 2a1aa27d..eb63f1d6 100644 --- a/server/gti/gti_mcp/tools/collections.py +++ b/server/gti/gti_mcp/tools/collections.py @@ -54,7 +54,7 @@ } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_collection_report(id: str, ctx: Context) -> typing.Dict[str, typing.Any]: """At Google Threat Intelligence, threats are modeled as "collections". This tool retrieves them from the platform. @@ -83,7 +83,7 @@ async def get_collection_report(id: str, ctx: Context) -> typing.Dict[str, typin return res -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_entities_related_to_a_collection( id: str, relationship_name: str, ctx: Context, limit: int = 10, descriptors_only: bool = True ) -> typing.List[typing.Dict[str, typing.Any]]: @@ -173,7 +173,7 @@ async def _search_threats_by_collection_type( return utils.sanitize_response([o.to_dict() for o in res]) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_threats( ctx: Context, query: str, @@ -236,7 +236,7 @@ async def search_threats( return utils.sanitize_response([o.to_dict() for o in res]) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_campaigns( query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-" ) -> typing.List[typing.Dict[str, typing.Any]]: @@ -259,7 +259,7 @@ async def search_campaigns( return res -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_threat_actors( query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-" ) -> typing.List[typing.Dict[str, typing.Any]]: @@ -282,7 +282,7 @@ async def search_threat_actors( return res -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_malware_families( query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-" ) -> typing.List[typing.Dict[str, typing.Any]]: @@ -305,7 +305,7 @@ async def search_malware_families( return res -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_software_toolkits( query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-" ) -> typing.List[typing.Dict[str, typing.Any]]: @@ -328,7 +328,7 @@ async def search_software_toolkits( return res -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_threat_reports( query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-" ) -> typing.List[typing.Dict[str, typing.Any]]: @@ -353,7 +353,7 @@ async def search_threat_reports( return res -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_vulnerabilities( query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-" ) -> typing.List[typing.Dict[str, typing.Any]]: @@ -376,7 +376,7 @@ async def search_vulnerabilities( return res -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_collection_timeline_events(id: str, ctx: Context): """Retrieves timeline events from the given collection, when available. @@ -401,7 +401,7 @@ async def get_collection_timeline_events(id: str, ctx: Context): return utils.sanitize_response(data.get("data", [])) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_collection_mitre_tree(id: str, ctx: Context) -> typing.Dict: """Retrieves the Mitre tactics and techniques associated with a threat. @@ -420,7 +420,12 @@ async def get_collection_mitre_tree(id: str, ctx: Context) -> typing.Dict: return utils.sanitize_response(data.get("data", {})) -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def create_collection( name: str, description: str, @@ -455,7 +460,12 @@ async def create_collection( return utils.sanitize_response(data["data"]) -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def update_collection_attributes( id: str, ctx: Context, @@ -483,7 +493,12 @@ async def update_collection_attributes( return utils.sanitize_response(data["data"]) -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def update_iocs_in_collection( id: str, ctx: Context, @@ -535,7 +550,7 @@ async def update_iocs_in_collection( return 'Sucesssfully updated collection' if status == 200 else 'Error updating collection' -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_collection_feature_matches( collection_id: str, feature_type: str, @@ -636,7 +651,7 @@ async def get_collection_feature_matches( return utils.sanitize_response(data.get("data", [])) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_collections_commonalities(collection_id: str, ctx: Context) -> str: """Retrieve the common characteristics or features (attributes / relationships) of the indicators of compromise (IoC) within a collection, identified by its ID. Args: @@ -706,7 +721,7 @@ async def _get_sigma_rule_details(ctx: Context, rule: dict, rule_type: str) -> t logging.exception("Error fetching Sigma ruleset %s: %s", ruleset_id, e) return {"error": f"Error fetching Sigma ruleset {ruleset_id}: {e}"} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_collection_rules(collection_id: str, ctx: Context, top_n: int = 4, rule_types: typing.List[str] = None) -> typing.Union[typing.List[typing.Dict[str, typing.Any]], typing.Dict[str, str]]: """Retrieve top N community rules and all curated hunting rules for a specific collection. diff --git a/server/gti/gti_mcp/tools/files.py b/server/gti/gti_mcp/tools/files.py index 6c7c6650..f4331ea0 100644 --- a/server/gti/gti_mcp/tools/files.py +++ b/server/gti/gti_mcp/tools/files.py @@ -85,7 +85,7 @@ ] -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_file_report(hash: str, ctx: Context) -> typing.Dict[str, typing.Any]: """Get a comprehensive file analysis report using its hash (MD5/SHA-1/SHA-256). @@ -107,7 +107,7 @@ async def get_file_report(hash: str, ctx: Context) -> typing.Dict[str, typing.An return utils.sanitize_response(res) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_entities_related_to_a_file( hash: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10, ) -> list[dict[str, typing.Any]]: @@ -190,7 +190,7 @@ async def get_entities_related_to_a_file( return utils.sanitize_response(res.get(relationship_name, [])) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_file_behavior_report( file_behaviour_id: str, ctx: Context ) -> typing.Dict[str, typing.Any]: @@ -225,7 +225,7 @@ async def get_file_behavior_report( return utils.sanitize_response(res) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_file_behavior_summary(hash: str, ctx: Context) -> typing.Dict[str, typing.Any]: """Retrieve a summary of all the file behavior reports from all the sandboxes. @@ -248,7 +248,12 @@ async def get_file_behavior_summary(hash: str, ctx: Context) -> typing.Dict[str, return utils.sanitize_response(res["data"]) -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def analyse_file(file_path: str, ctx: Context): """Upload and analyse the file in VirusTotal. @@ -268,7 +273,7 @@ async def analyse_file(file_path: str, ctx: Context): logging.info(f"Analysis has completed with ID %s", res.id) return utils.sanitize_response(res.to_dict()) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_digital_threat_monitoring( query: str, ctx: Context, diff --git a/server/gti/gti_mcp/tools/intelligence.py b/server/gti/gti_mcp/tools/intelligence.py index 9e336813..2b5298fa 100644 --- a/server/gti/gti_mcp/tools/intelligence.py +++ b/server/gti/gti_mcp/tools/intelligence.py @@ -24,7 +24,7 @@ ] -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_iocs(query: str, ctx: Context, limit: int = 10, order_by: str = "last_submission_date-") -> typing.List[typing.Dict[str, typing.Any]]: """Search Indicators of Compromise (IOC) in the Google Threat Intelligence platform. @@ -69,7 +69,7 @@ async def search_iocs(query: str, ctx: Context, limit: int = 10, order_by: str = return utils.sanitize_response([o.to_dict() for o in res]) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_hunting_ruleset(ruleset_id: str, ctx: Context) -> typing.Dict[str, typing.Any]: """Get a Hunting Ruleset object from Google Threat Intelligence. @@ -103,7 +103,7 @@ async def get_hunting_ruleset(ruleset_id: str, ctx: Context) -> typing.Dict[str, return utils.sanitize_response(res) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_entities_related_to_a_hunting_ruleset( ruleset_id: str, relationship_name: str, ctx: Context, limit: int = 10 ) -> list[dict[str, typing.Any]]: diff --git a/server/gti/gti_mcp/tools/netloc.py b/server/gti/gti_mcp/tools/netloc.py index ceda4a54..28a053c9 100644 --- a/server/gti/gti_mcp/tools/netloc.py +++ b/server/gti/gti_mcp/tools/netloc.py @@ -88,7 +88,7 @@ ] -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_domain_report(domain: str, ctx: Context) -> typing.Dict[str, typing.Any]: """Get a comprehensive domain analysis report from Google Threat Intelligence. @@ -108,7 +108,7 @@ async def get_domain_report(domain: str, ctx: Context) -> typing.Dict[str, typin return utils.sanitize_response(res) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_entities_related_to_a_domain( domain: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10 ) -> list[dict[str, typing.Any]]: @@ -174,7 +174,7 @@ async def get_entities_related_to_a_domain( return utils.sanitize_response(res.get(relationship_name, [])) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_ip_address_report(ip_address: str, ctx: Context) -> typing.Dict[str, typing.Any]: """Get a comprehensive IP Address analysis report from Google Threat Intelligence. @@ -193,7 +193,7 @@ async def get_ip_address_report(ip_address: str, ctx: Context) -> typing.Dict[st return utils.sanitize_response(res) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_entities_related_to_an_ip_address( ip_address: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10 ) -> list[dict[str, typing.Any]]: diff --git a/server/gti/gti_mcp/tools/threat_profiles.py b/server/gti/gti_mcp/tools/threat_profiles.py index be3486b1..e67bde98 100644 --- a/server/gti/gti_mcp/tools/threat_profiles.py +++ b/server/gti/gti_mcp/tools/threat_profiles.py @@ -19,7 +19,7 @@ from ..server import server, vt_client -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_threat_profiles( ctx: Context, limit: int = 10 ) -> typing.List[typing.Dict[str, typing.Any]]: @@ -51,7 +51,7 @@ async def list_threat_profiles( return utils.sanitize_response([o.to_dict() for o in res]) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_threat_profile( profile_id: str, ctx: Context ) -> typing.Dict[str, typing.Any]: @@ -99,7 +99,7 @@ async def get_threat_profile( return utils.sanitize_response(res) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_threat_profile_recommendations( profile_id: str, ctx: Context, limit: int = 10 ) -> typing.List[typing.Dict[str, typing.Any]]: @@ -144,7 +144,7 @@ async def get_threat_profile_recommendations( return utils.sanitize_response(res.get('recommendations', [])) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_threat_profile_associations_timeline( profile_id: str, ctx: Context, limit: int = 10 ) -> typing.List[typing.Dict[str, typing.Any]]: diff --git a/server/gti/gti_mcp/tools/urls.py b/server/gti/gti_mcp/tools/urls.py index e88f5254..483de929 100644 --- a/server/gti/gti_mcp/tools/urls.py +++ b/server/gti/gti_mcp/tools/urls.py @@ -69,7 +69,7 @@ def url_to_base64(url: str) -> str: return b.decode('utf-8').rstrip("=") -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_url_report(url: str, ctx: Context) -> typing.Dict[str, typing.Any]: """Get a comprehensive URL analysis report from Google Threat Intelligence. @@ -90,7 +90,7 @@ async def get_url_report(url: str, ctx: Context) -> typing.Dict[str, typing.Any] return utils.sanitize_response(res) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_entities_related_to_an_url( url: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10 ) -> list[dict[str, typing.Any]]: diff --git a/server/gti/tests/test_tool_annotations_unit.py b/server/gti/tests/test_tool_annotations_unit.py new file mode 100644 index 00000000..ae57dd29 --- /dev/null +++ b/server/gti/tests/test_tool_annotations_unit.py @@ -0,0 +1,123 @@ +# Copyright 2025 Google LLC +# +# 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. +"""Unit tests for FastMCP tool annotations in gti-mcp.""" + +import pytest +from mcp.types import ToolAnnotations + +import gti_mcp.server as gti_server + +MUTATING_ADDITIVE_TOOLS = [ + "create_collection", + "update_collection_attributes", + "update_iocs_in_collection", + "analyse_file", +] + +READ_ONLY_TOOLS = [ + "get_collection_report", + "get_entities_related_to_a_collection", + "search_threats", + "search_campaigns", + "search_threat_actors", + "search_malware_families", + "search_software_toolkits", + "search_threat_reports", + "search_vulnerabilities", + "get_collection_timeline_events", + "get_collection_mitre_tree", + "get_collection_feature_matches", + "get_collections_commonalities", + "get_collection_rules", + "get_file_report", + "get_entities_related_to_a_file", + "get_file_behavior_report", + "get_file_behavior_summary", + "search_digital_threat_monitoring", + "search_iocs", + "get_hunting_ruleset", + "get_entities_related_to_a_hunting_ruleset", + "get_domain_report", + "get_entities_related_to_a_domain", + "get_ip_address_report", + "get_entities_related_to_an_ip_address", + "list_threat_profiles", + "get_threat_profile", + "get_threat_profile_recommendations", + "get_threat_profile_associations_timeline", + "get_url_report", + "get_entities_related_to_an_url", +] + + +def test_tool_catalog_coverage(): + """Verify our test suites cover all 36 tools registered on the server.""" + tools = gti_server.server._tool_manager.list_tools() + all_registered_names = {t.name for t in tools} + tested_names = set(MUTATING_ADDITIVE_TOOLS) | set(READ_ONLY_TOOLS) + + assert all_registered_names == tested_names, ( + f"Mismatch in tested tools vs registered tools. " + f"Missing from tests: {all_registered_names - tested_names}. " + f"Extra in tests: {tested_names - all_registered_names}" + ) + + +def test_all_tools_have_annotations(): + """Verify every registered tool defines explicit ToolAnnotations with readOnlyHint.""" + tools = gti_server.server._tool_manager.list_tools() + assert len(tools) > 0, "No tools registered on server" + + missing_annotations = [] + missing_hints = [] + + for tool in tools: + if tool.annotations is None: + missing_annotations.append(tool.name) + continue + + assert isinstance(tool.annotations, ToolAnnotations) + if tool.annotations.readOnlyHint is None: + missing_hints.append(f"{tool.name}: readOnlyHint is None") + + if tool.annotations.readOnlyHint is False and tool.annotations.destructiveHint is None: + missing_hints.append(f"{tool.name}: destructiveHint is None for mutating tool") + + assert not missing_annotations, ( + f"The following {len(missing_annotations)} tools are missing annotations: " + f"{', '.join(sorted(missing_annotations))}" + ) + assert not missing_hints, ( + f"The following tools have incomplete hint definitions: " + f"{', '.join(sorted(missing_hints))}" + ) + + +@pytest.mark.parametrize("tool_name", MUTATING_ADDITIVE_TOOLS) +def test_mutating_additive_tool_annotations(tool_name: str): + """Verify additive mutating tools declare readOnlyHint=False and destructiveHint=False.""" + tool = gti_server.server._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is False, f"Expected {tool_name} readOnlyHint=False" + assert tool.annotations.destructiveHint is False, f"Expected {tool_name} destructiveHint=False" + + +@pytest.mark.parametrize("tool_name", READ_ONLY_TOOLS) +def test_read_only_tool_annotations(tool_name: str): + """Verify query and search tools declare readOnlyHint=True.""" + tool = gti_server.server._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is True, f"Expected {tool_name} readOnlyHint=True" diff --git a/server/scc/README.md b/server/scc/README.md index da8264cf..7f787f16 100644 --- a/server/scc/README.md +++ b/server/scc/README.md @@ -4,6 +4,13 @@ This is an MCP (Model Context Protocol) server for interacting with Google Cloud ## Features +### Tool Annotations & Safety Hints + +All tools exposed by the Security Command Center MCP server declare Model Context Protocol `ToolAnnotations` metadata to guide autonomous clients (such as Gemini CLI, Google ADK, Claude, and Cursor) regarding execution safety: + +- **Read-Only Tools (`readOnlyHint: true`)**: Finding search, finding details inspection, compliance evaluation, vulnerability ranking, and remediation step retrieval (`search_findings`, `get_finding_details`, `search_findings_by_compliance`, `top_vulnerability_findings`, `get_finding_remediation`). +- **Additive / Mutating Tools (`readOnlyHint: false, destructiveHint: false`)**: Operations that mutate finding states without deleting data (`set_finding_mute`). + ### Available Tools - **`search_findings(project_id, finding_class=None, severity=None, state="ACTIVE", category=None, ...)`** diff --git a/server/scc/scc_mcp.py b/server/scc/scc_mcp.py index 1869967e..e1623f67 100644 --- a/server/scc/scc_mcp.py +++ b/server/scc/scc_mcp.py @@ -97,7 +97,7 @@ def _build_or_filter(field: str, value: str) -> str: # --- Security Command Center Tools --- -@mcp.tool() +@mcp.tool(annotations={"readOnlyHint": True}) async def search_findings( project_id: str = None, organization_id: str = None, @@ -222,7 +222,7 @@ async def search_findings( return {"error": "An unexpected error occurred", "details": str(e)} -@mcp.tool() +@mcp.tool(annotations={"readOnlyHint": True}) async def get_finding_details( project_id: str = None, organization_id: str = None, @@ -331,7 +331,7 @@ async def get_finding_details( return {"error": "An unexpected error occurred", "details": str(e)} -@mcp.tool() +@mcp.tool(annotations={"readOnlyHint": True}) async def search_findings_by_compliance( project_id: str = None, organization_id: str = None, @@ -517,7 +517,7 @@ async def search_findings_by_compliance( return {"error": "An unexpected error occurred", "details": str(e)} -@mcp.tool() +@mcp.tool(annotations={"readOnlyHint": True}) async def top_vulnerability_findings( project_id: str = None, organization_id: str = None, @@ -610,7 +610,7 @@ async def top_vulnerability_findings( return {"error": "An unexpected error occurred", "details": str(e)} -@mcp.tool() +@mcp.tool(annotations={"readOnlyHint": True}) async def get_finding_remediation( project_id: str = None, organization_id: str = None, @@ -771,7 +771,12 @@ async def get_finding_remediation( return {"error": "An unexpected error occurred", "details": str(e)} -@mcp.tool() +@mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def set_finding_mute( project_id: str = None, organization_id: str = None, diff --git a/server/scc/tests/test_tool_annotations_unit.py b/server/scc/tests/test_tool_annotations_unit.py new file mode 100644 index 00000000..91a2a8b2 --- /dev/null +++ b/server/scc/tests/test_tool_annotations_unit.py @@ -0,0 +1,96 @@ +# Copyright 2025 Google LLC +# +# 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. +"""Unit tests for FastMCP tool annotations in scc-mcp.""" + +import os +os.environ["GOOGLE_API_USE_CLIENT_CERTIFICATE"] = "false" + +import pytest +from mcp.types import ToolAnnotations + +import scc_mcp + +MUTATING_ADDITIVE_TOOLS = [ + "set_finding_mute", +] + +READ_ONLY_TOOLS = [ + "search_findings", + "get_finding_details", + "search_findings_by_compliance", + "top_vulnerability_findings", + "get_finding_remediation", +] + + +def test_tool_catalog_coverage(): + """Verify our test suites cover all 6 tools registered on the server.""" + tools = scc_mcp.mcp._tool_manager.list_tools() + all_registered_names = {t.name for t in tools} + tested_names = set(MUTATING_ADDITIVE_TOOLS) | set(READ_ONLY_TOOLS) + + assert all_registered_names == tested_names, ( + f"Mismatch in tested tools vs registered tools. " + f"Missing from tests: {all_registered_names - tested_names}. " + f"Extra in tests: {tested_names - all_registered_names}" + ) + + +def test_all_tools_have_annotations(): + """Verify every registered tool defines explicit ToolAnnotations with readOnlyHint.""" + tools = scc_mcp.mcp._tool_manager.list_tools() + assert len(tools) > 0, "No tools registered on server" + + missing_annotations = [] + missing_hints = [] + + for tool in tools: + if tool.annotations is None: + missing_annotations.append(tool.name) + continue + + assert isinstance(tool.annotations, ToolAnnotations) + if tool.annotations.readOnlyHint is None: + missing_hints.append(f"{tool.name}: readOnlyHint is None") + + if tool.annotations.readOnlyHint is False and tool.annotations.destructiveHint is None: + missing_hints.append(f"{tool.name}: destructiveHint is None for mutating tool") + + assert not missing_annotations, ( + f"The following {len(missing_annotations)} tools are missing annotations: " + f"{', '.join(sorted(missing_annotations))}" + ) + assert not missing_hints, ( + f"The following tools have incomplete hint definitions: " + f"{', '.join(sorted(missing_hints))}" + ) + + +@pytest.mark.parametrize("tool_name", MUTATING_ADDITIVE_TOOLS) +def test_mutating_additive_tool_annotations(tool_name: str): + """Verify additive mutating tools declare readOnlyHint=False and destructiveHint=False.""" + tool = scc_mcp.mcp._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is False, f"Expected {tool_name} readOnlyHint=False" + assert tool.annotations.destructiveHint is False, f"Expected {tool_name} destructiveHint=False" + + +@pytest.mark.parametrize("tool_name", READ_ONLY_TOOLS) +def test_read_only_tool_annotations(tool_name: str): + """Verify query and search tools declare readOnlyHint=True.""" + tool = scc_mcp.mcp._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is True, f"Expected {tool_name} readOnlyHint=True" diff --git a/server/secops-soar/README.md b/server/secops-soar/README.md index 96c2f7b6..fba20488 100644 --- a/server/secops-soar/README.md +++ b/server/secops-soar/README.md @@ -6,6 +6,14 @@ Chronicle Security Operations SOAR suite. ## Features +### Tool Annotations & Safety Hints + +All case management tools exposed by the Chronicle SecOps SOAR MCP server declare Model Context Protocol `ToolAnnotations` metadata to guide autonomous clients (such as Gemini CLI, Google ADK, Claude, and Cursor) regarding execution safety: + +- **Read-Only Tools (`readOnlyHint: true`)**: Case listing, alert retrieval, event inspection, entity details, entity searches, and full case detail queries (`list_cases`, `list_alerts_by_case`, `list_alert_group_identifiers_by_case`, `list_events_by_alert`, `get_entities_by_alert_group_identifiers`, `get_entity_details`, `search_entity`, `get_case_full_details`). +- **Additive / Mutating Tools (`readOnlyHint: false, destructiveHint: false`)**: Operations that create cases, post comments, update priority/description, assign analysts, transition stage, or attach tags (`create_case`, `post_case_comment`, `change_case_priority`, `update_case_description`, `assign_case`, `change_case_stage`, `add_case_tag`). +- **Destructive Tools (`readOnlyHint: false, destructiveHint: true`)**: Operations that conclude incident lifecycles or remove tags (`close_case`, `remove_case_tag`). + ### Core Tools (Case Management & Entities) - **`list_cases()`** - Lists available cases in the SOAR platform. diff --git a/server/secops-soar/secops_soar_mcp/case_management.py b/server/secops-soar/secops_soar_mcp/case_management.py index cb7444c6..5717a4ba 100644 --- a/server/secops-soar/secops_soar_mcp/case_management.py +++ b/server/secops-soar/secops_soar_mcp/case_management.py @@ -25,7 +25,12 @@ def register_tools(mcp: FastMCP): - @mcp.tool() + @mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } + ) async def create_case( name: Annotated[str, Field(..., description="The name or title of the case.")], priority: Annotated[ @@ -107,7 +112,7 @@ async def create_case( req["Environment"] = environment return await bindings.http_client.post(Endpoints.BASE_CASE_URL, req=req) - @mcp.tool() + @mcp.tool(annotations={"readOnlyHint": True}) async def list_cases( next_page_token: Annotated[ Optional[str], @@ -148,7 +153,12 @@ async def list_cases( ) return await bindings.http_client.get(Endpoints.BASE_CASE_URL) - @mcp.tool() + @mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } + ) async def post_case_comment( case_id: Annotated[str, Field(..., description="The ID of the case.")], comment: Annotated[ @@ -187,7 +197,7 @@ async def post_case_comment( req={"Comment": comment}, ) - @mcp.tool() + @mcp.tool(annotations={"readOnlyHint": True}) async def list_alerts_by_case( case_id: Annotated[str, Field(..., description="The ID of the case.")], next_page_token: Annotated[ @@ -237,7 +247,7 @@ async def list_alerts_by_case( Endpoints.BASE_ALERT_URL.format(CASE_ID=case_id) ) - @mcp.tool() + @mcp.tool(annotations={"readOnlyHint": True}) async def list_alert_group_identifiers_by_case( case_id: Annotated[str, Field(..., description="The ID of the case.")], next_page_token: Annotated[ @@ -285,7 +295,7 @@ async def list_alert_group_identifiers_by_case( Endpoints.LIST_ALERT_GROUP_IDENTIFIERS_BY_CASE.format(CASE_ID=case_id) ) - @mcp.tool() + @mcp.tool(annotations={"readOnlyHint": True}) async def list_events_by_alert( case_id: Annotated[str, Field(..., description="The ID of the case.")], alert_id: Annotated[str, Field(..., description="The ID of the alert.")], @@ -340,7 +350,12 @@ async def list_events_by_alert( ) ) - @mcp.tool() + @mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } + ) async def change_case_priority( case_id: Annotated[str, Field(..., description="The ID of the case.")], case_priority: Annotated[ @@ -394,7 +409,12 @@ async def change_case_priority( req={"Priority": case_priority}, ) - @mcp.tool() + @mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } + ) async def update_case_description( case_id: Annotated[str, Field(..., description="The ID of the case.")], description: Annotated[ @@ -445,7 +465,12 @@ async def update_case_description( req={"CaseId": case_id, "Description": description}, ) - @mcp.tool() + @mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": True, + } + ) async def close_case( case_id: Annotated[str, Field(..., description="The ID of the case.")], root_cause: Annotated[ @@ -527,7 +552,12 @@ async def close_case( req["Tags"] = tags return await bindings.http_client.post(Endpoints.CLOSE_CASE, req=req) - @mcp.tool() + @mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } + ) async def assign_case( case_id: Annotated[str, Field(..., description="The ID of the case.")], user: Annotated[ @@ -571,7 +601,12 @@ async def assign_case( req={"CaseId": case_id, "User": user}, ) - @mcp.tool() + @mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } + ) async def change_case_stage( case_id: Annotated[str, Field(..., description="The ID of the case.")], stage: Annotated[ @@ -617,7 +652,12 @@ async def change_case_stage( req={"CaseId": case_id, "Stage": stage}, ) - @mcp.tool() + @mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } + ) async def add_case_tag( case_id: Annotated[str, Field(..., description="The ID of the case.")], tag: Annotated[ @@ -662,7 +702,12 @@ async def add_case_tag( req={"CaseId": case_id, "Tag": tag}, ) - @mcp.tool() + @mcp.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": True, + } + ) async def remove_case_tag( case_id: Annotated[str, Field(..., description="The ID of the case.")], tag: Annotated[ @@ -705,7 +750,7 @@ async def remove_case_tag( req={"CaseId": case_id, "Tag": tag}, ) - @mcp.tool() + @mcp.tool(annotations={"readOnlyHint": True}) async def get_entities_by_alert_group_identifiers( case_id: Annotated[str, Field(..., description="The ID of the case.")], alert_group_identifiers: Annotated[ @@ -749,7 +794,7 @@ async def get_entities_by_alert_group_identifiers( req={"caseId": case_id, "alertGroupIdentifiers": alert_group_identifiers}, ) - @mcp.tool() + @mcp.tool(annotations={"readOnlyHint": True}) async def get_entity_details( entity_identifier: Annotated[ str, Field(..., description="The identifier of the entity.") @@ -802,7 +847,7 @@ async def get_entity_details( }, ) - @mcp.tool() + @mcp.tool(annotations={"readOnlyHint": True}) async def search_entity( term: Annotated[ Optional[str], @@ -902,7 +947,7 @@ async def search_entity( }, ) - @mcp.tool() + @mcp.tool(annotations={"readOnlyHint": True}) async def get_case_full_details( case_id: Annotated[str, Field(..., description="The ID of the case.")], ): diff --git a/server/secops-soar/tests/unit/test_tool_annotations_unit.py b/server/secops-soar/tests/unit/test_tool_annotations_unit.py new file mode 100644 index 00000000..3b57bb0b --- /dev/null +++ b/server/secops-soar/tests/unit/test_tool_annotations_unit.py @@ -0,0 +1,117 @@ +# Copyright 2025 Google LLC +# +# 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. +"""Unit tests for FastMCP tool annotations in secops-soar-mcp.""" + +import pytest +from mcp.types import ToolAnnotations + +from secops_soar_mcp.server import mcp + +DESTRUCTIVE_TOOLS = [ + "close_case", + "remove_case_tag", +] + +MUTATING_ADDITIVE_TOOLS = [ + "create_case", + "post_case_comment", + "change_case_priority", + "update_case_description", + "assign_case", + "change_case_stage", + "add_case_tag", +] + +READ_ONLY_TOOLS = [ + "list_cases", + "list_alerts_by_case", + "list_alert_group_identifiers_by_case", + "list_events_by_alert", + "get_entities_by_alert_group_identifiers", + "get_entity_details", + "search_entity", + "get_case_full_details", +] + + +def test_tool_catalog_coverage(): + """Verify our test suites cover all 17 case management tools registered on the server.""" + tools = mcp._tool_manager.list_tools() + all_registered_names = {t.name for t in tools} + tested_names = set(DESTRUCTIVE_TOOLS) | set(MUTATING_ADDITIVE_TOOLS) | set(READ_ONLY_TOOLS) + + assert all_registered_names == tested_names, ( + f"Mismatch in tested tools vs registered tools. " + f"Missing from tests: {all_registered_names - tested_names}. " + f"Extra in tests: {tested_names - all_registered_names}" + ) + + +def test_all_tools_have_annotations(): + """Verify every registered tool defines explicit ToolAnnotations with readOnlyHint.""" + tools = mcp._tool_manager.list_tools() + assert len(tools) > 0, "No tools registered on server" + + missing_annotations = [] + missing_hints = [] + + for tool in tools: + if tool.annotations is None: + missing_annotations.append(tool.name) + continue + + assert isinstance(tool.annotations, ToolAnnotations) + if tool.annotations.readOnlyHint is None: + missing_hints.append(f"{tool.name}: readOnlyHint is None") + + if tool.annotations.readOnlyHint is False and tool.annotations.destructiveHint is None: + missing_hints.append(f"{tool.name}: destructiveHint is None for mutating tool") + + assert not missing_annotations, ( + f"The following {len(missing_annotations)} tools are missing annotations: " + f"{', '.join(sorted(missing_annotations))}" + ) + assert not missing_hints, ( + f"The following tools have incomplete hint definitions: " + f"{', '.join(sorted(missing_hints))}" + ) + + +@pytest.mark.parametrize("tool_name", DESTRUCTIVE_TOOLS) +def test_destructive_tool_annotations(tool_name: str): + """Verify destructive tools declare readOnlyHint=False and destructiveHint=True.""" + tool = mcp._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is False, f"Expected {tool_name} readOnlyHint=False" + assert tool.annotations.destructiveHint is True, f"Expected {tool_name} destructiveHint=True" + + +@pytest.mark.parametrize("tool_name", MUTATING_ADDITIVE_TOOLS) +def test_mutating_additive_tool_annotations(tool_name: str): + """Verify additive mutating tools declare readOnlyHint=False and destructiveHint=False.""" + tool = mcp._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is False, f"Expected {tool_name} readOnlyHint=False" + assert tool.annotations.destructiveHint is False, f"Expected {tool_name} destructiveHint=False" + + +@pytest.mark.parametrize("tool_name", READ_ONLY_TOOLS) +def test_read_only_tool_annotations(tool_name: str): + """Verify query and search tools declare readOnlyHint=True.""" + tool = mcp._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is True, f"Expected {tool_name} readOnlyHint=True" diff --git a/server/secops/README.md b/server/secops/README.md index 8b1d7600..bafacc79 100644 --- a/server/secops/README.md +++ b/server/secops/README.md @@ -6,6 +6,14 @@ Chronicle Security Operations suite. ## Features +### Tool Annotations & Safety Hints + +All tools exposed by the Chronicle SecOps MCP server declare Model Context Protocol `ToolAnnotations` metadata to guide autonomous clients (such as Gemini CLI, Google ADK, Claude, and Cursor) regarding execution safety: + +- **Read-Only Tools (`readOnlyHint: true`)**: Telemetry searches, entity lookups, alert retrieval, parser and rule validations, and listing operations that query Chronicle without modifying state. Clients can execute these tools safely without interactive prompts. +- **Additive / Mutating Tools (`readOnlyHint: false, destructiveHint: false`)**: Operations that create or update resources (such as `create_rule`, `create_data_table`, `add_rows_to_data_table`, `ingest_raw_log`, `create_feed`) without deleting or invalidating data. +- **Destructive Tools (`readOnlyHint: false, destructiveHint: true`)**: High-impact operations that permanently remove resources or disrupt data collection (such as `delete_data_table_rows`, `delete_feed`, `delete_watchlist`, `disable_feed`, `generate_feed_secret`, `deactivate_parser`). Clients can use this hint to require explicit user confirmation before executing these actions. + ### Security Tools - **`search_security_events(text, project_id=None, customer_id=None, hours_back=24, max_events=100, region=None)`** diff --git a/server/secops/secops_mcp/tools/curated_rules_management.py b/server/secops/secops_mcp/tools/curated_rules_management.py index 62da88f2..eace5188 100644 --- a/server/secops/secops_mcp/tools/curated_rules_management.py +++ b/server/secops/secops_mcp/tools/curated_rules_management.py @@ -23,7 +23,7 @@ logger = logging.getLogger("secops-mcp") -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_curated_rules( project_id: Optional[str] = None, customer_id: Optional[str] = None, @@ -105,7 +105,7 @@ async def list_curated_rules( return {"error": str(e), "curatedRules": []} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_curated_rule( rule_id: str, project_id: Optional[str] = None, @@ -175,7 +175,7 @@ async def get_curated_rule( } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_curated_rule_by_name( display_name: str, project_id: Optional[str] = None, @@ -243,7 +243,7 @@ async def get_curated_rule_by_name( } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_curated_detections( rule_id: str, start_time: str, @@ -357,7 +357,7 @@ async def search_curated_detections( } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_curated_rule_sets( project_id: Optional[str] = None, customer_id: Optional[str] = None, @@ -445,7 +445,7 @@ async def list_curated_rule_sets( return {"error": str(e), "curatedRuleSets": []} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_curated_rule_set( rule_set_id: str, project_id: Optional[str] = None, @@ -516,7 +516,7 @@ async def get_curated_rule_set( } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_curated_rule_set_deployments( project_id: Optional[str] = None, customer_id: Optional[str] = None, @@ -608,7 +608,12 @@ async def list_curated_rule_set_deployments( return {"error": str(e), "curatedRuleSetDeployments": []} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def update_curated_rule_set_deployment( category_id: str, rule_set_id: str, diff --git a/server/secops/secops_mcp/tools/data_table_management.py b/server/secops/secops_mcp/tools/data_table_management.py index a1b14939..dec5253f 100644 --- a/server/secops/secops_mcp/tools/data_table_management.py +++ b/server/secops/secops_mcp/tools/data_table_management.py @@ -22,7 +22,12 @@ # Configure logging logger = logging.getLogger('secops-mcp') -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def create_data_table( name: str, description: str, @@ -161,7 +166,12 @@ async def create_data_table( logger.error(f'Error creating data table {name}: {str(e)}', exc_info=True) return f'Error creating data table {name}: {str(e)}' -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def add_rows_to_data_table( table_name: str, rows: List[List[str]], @@ -270,7 +280,7 @@ async def add_rows_to_data_table( logger.error(f'Error adding rows to data table {table_name}: {str(e)}', exc_info=True) return f'Error adding rows to data table {table_name}: {str(e)}' -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_data_table_rows( table_name: str, project_id: Optional[str] = None, @@ -373,7 +383,12 @@ async def list_data_table_rows( logger.error(f'Error listing rows in data table {table_name}: {str(e)}', exc_info=True) return f'Error listing rows in data table {table_name}: {str(e)}' -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": True, + } +) async def delete_data_table_rows( table_name: str, row_ids: List[str], diff --git a/server/secops/secops_mcp/tools/entity_lookup.py b/server/secops/secops_mcp/tools/entity_lookup.py index b8238db4..fe8233af 100644 --- a/server/secops/secops_mcp/tools/entity_lookup.py +++ b/server/secops/secops_mcp/tools/entity_lookup.py @@ -23,7 +23,7 @@ # Configure logging logger = logging.getLogger('secops-mcp') -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def lookup_entity( entity_value: str, project_id: Optional[str] = None, diff --git a/server/secops/secops_mcp/tools/feed_management.py b/server/secops/secops_mcp/tools/feed_management.py index 22dcc644..b77eb4f9 100644 --- a/server/secops/secops_mcp/tools/feed_management.py +++ b/server/secops/secops_mcp/tools/feed_management.py @@ -23,7 +23,7 @@ logger = logging.getLogger("secops-mcp") -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_feeds( project_id: Optional[str] = None, customer_id: Optional[str] = None, @@ -102,7 +102,7 @@ async def list_feeds( return {"error": f"Error listing feeds: {e}"} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_feed( feed_id: str, project_id: Optional[str] = None, @@ -166,7 +166,12 @@ async def get_feed( return {"error": f"Error getting feed: {e}"} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def create_feed( display_name: str, feed_details: Dict[str, Any], @@ -242,7 +247,12 @@ async def create_feed( return {"error": f"Error creating feed: {e}"} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def update_feed( feed_id: str, display_name: Optional[str] = None, @@ -317,7 +327,12 @@ async def update_feed( return {"error": f"Error updating feed: {e}"} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def enable_feed( feed_id: str, project_id: Optional[str] = None, @@ -382,7 +397,12 @@ async def enable_feed( return {"error": f"Error enabling feed: {e}"} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": True, + } +) async def disable_feed( feed_id: str, project_id: Optional[str] = None, @@ -447,7 +467,12 @@ async def disable_feed( return {"error": f"Error disabling feed: {e}"} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": True, + } +) async def delete_feed( feed_id: str, project_id: Optional[str] = None, @@ -508,7 +533,12 @@ async def delete_feed( return {"error": f"Error deleting feed: {e}"} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": True, + } +) async def generate_feed_secret( feed_id: str, project_id: Optional[str] = None, diff --git a/server/secops/secops_mcp/tools/investigation_management.py b/server/secops/secops_mcp/tools/investigation_management.py index f419eee0..528dbce7 100644 --- a/server/secops/secops_mcp/tools/investigation_management.py +++ b/server/secops/secops_mcp/tools/investigation_management.py @@ -22,7 +22,7 @@ logger = logging.getLogger("secops-mcp") -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_investigations( page_size: int = 50, page_token: Optional[str] = None, @@ -92,7 +92,7 @@ async def list_investigations( return {"error": error_msg} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_investigation( investigation_id: str, project_id: Optional[str] = None, @@ -172,7 +172,12 @@ async def get_investigation( return {"error": error_msg} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def trigger_investigation( alert_id: str, project_id: Optional[str] = None, @@ -261,7 +266,7 @@ async def trigger_investigation( return {"error": error_msg} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def fetch_associated_investigations( detection_type: str, alert_ids: Optional[List[str]] = None, diff --git a/server/secops/secops_mcp/tools/ioc_matches.py b/server/secops/secops_mcp/tools/ioc_matches.py index bfd99d15..119224f2 100644 --- a/server/secops/secops_mcp/tools/ioc_matches.py +++ b/server/secops/secops_mcp/tools/ioc_matches.py @@ -23,7 +23,7 @@ # Configure logging logger = logging.getLogger('secops-mcp') -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_ioc_matches( project_id: Optional[str] = None, customer_id: Optional[str] = None, diff --git a/server/secops/secops_mcp/tools/log_ingestion.py b/server/secops/secops_mcp/tools/log_ingestion.py index 6fbc27ef..6ba27cab 100644 --- a/server/secops/secops_mcp/tools/log_ingestion.py +++ b/server/secops/secops_mcp/tools/log_ingestion.py @@ -25,7 +25,12 @@ # Configure logging logger = logging.getLogger('secops-mcp') -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def ingest_raw_log( log_type: str, log_message: Union[str, List[str]], @@ -142,7 +147,12 @@ async def ingest_raw_log( logger.error(f'Error ingesting raw log: {str(e)}', exc_info=True) return f'Error ingesting raw log: {str(e)}' -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def ingest_udm_events( udm_events: Union[Dict[str, Any], List[Dict[str, Any]]], project_id: Optional[str] = None, @@ -291,7 +301,7 @@ async def ingest_udm_events( logger.error(f'Error ingesting UDM events: {str(e)}', exc_info=True) return f'Error ingesting UDM events: {str(e)}' -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_available_log_types( project_id: Optional[str] = None, customer_id: Optional[str] = None, diff --git a/server/secops/secops_mcp/tools/parser_management.py b/server/secops/secops_mcp/tools/parser_management.py index a7df6670..6921fa92 100644 --- a/server/secops/secops_mcp/tools/parser_management.py +++ b/server/secops/secops_mcp/tools/parser_management.py @@ -25,7 +25,12 @@ logger = logging.getLogger("secops-mcp") -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def create_parser( log_type: str, parser_code: str, @@ -144,7 +149,7 @@ async def create_parser( return f"Error creating parser for log type {log_type}: {str(e)}" -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_parser( log_type: str, parser_id: str, @@ -242,7 +247,7 @@ async def get_parser( return f"Error getting parser {parser_id} for log type {log_type}: {str(e)}" -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_parsers( log_type: str = "-", page_size: Optional[int] = None, @@ -345,7 +350,12 @@ async def list_parsers( } -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def activate_parser( log_type: str, parser_id: str, @@ -424,7 +434,12 @@ async def activate_parser( return f"Error activating parser {parser_id} for log type {log_type}: {str(e)}" -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": True, + } +) async def deactivate_parser( log_type: str, parser_id: str, @@ -507,7 +522,7 @@ async def deactivate_parser( ) -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def run_parser_against_sample_logs( log_type: str, parser_code: str, diff --git a/server/secops/secops_mcp/tools/reference_list_management.py b/server/secops/secops_mcp/tools/reference_list_management.py index 20de0630..5a772bc1 100644 --- a/server/secops/secops_mcp/tools/reference_list_management.py +++ b/server/secops/secops_mcp/tools/reference_list_management.py @@ -22,7 +22,12 @@ # Configure logging logger = logging.getLogger('secops-mcp') -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def create_reference_list( name: str, description: str, @@ -154,7 +159,7 @@ async def create_reference_list( logger.error(f'Error creating reference list {name}: {str(e)}', exc_info=True) return f'Error creating reference list {name}: {str(e)}' -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_reference_list( name: str, project_id: Optional[str] = None, @@ -290,7 +295,12 @@ async def get_reference_list( logger.error(f'Error getting reference list {name}: {str(e)}', exc_info=True) return f'Error getting reference list {name}: {str(e)}' -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def update_reference_list( name: str, project_id: Optional[str] = None, diff --git a/server/secops/secops_mcp/tools/rule_exclusions.py b/server/secops/secops_mcp/tools/rule_exclusions.py index 97eb6b29..8b1b7c99 100644 --- a/server/secops/secops_mcp/tools/rule_exclusions.py +++ b/server/secops/secops_mcp/tools/rule_exclusions.py @@ -23,7 +23,12 @@ logger = logging.getLogger("secops-mcp") -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def create_rule_exclusion( display_name: str, refinement_type: str, @@ -127,7 +132,7 @@ async def create_rule_exclusion( } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_rule_exclusion( exclusion_id: str, project_id: Optional[str] = None, @@ -215,7 +220,7 @@ async def get_rule_exclusion( return {"error": error_msg} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_rule_exclusions( page_size: Optional[int] = None, page_token: Optional[str] = None, @@ -314,7 +319,12 @@ async def list_rule_exclusions( return {"error": error_msg} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def patch_rule_exclusion( exclusion_id: str, display_name: Optional[str] = None, @@ -416,7 +426,12 @@ async def patch_rule_exclusion( } -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def update_rule_exclusion_deployment( exclusion_id: str, enabled: bool, @@ -534,7 +549,7 @@ async def update_rule_exclusion_deployment( } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def compute_rule_exclusion_activity( exclusion_id: str, start_time: datetime, diff --git a/server/secops/secops_mcp/tools/search.py b/server/secops/secops_mcp/tools/search.py index a7c52c32..1a1de00d 100644 --- a/server/secops/secops_mcp/tools/search.py +++ b/server/secops/secops_mcp/tools/search.py @@ -24,7 +24,7 @@ # Configure logging logger = logging.getLogger('secops-mcp') -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_udm( query: str, hours_back: int = 24, diff --git a/server/secops/secops_mcp/tools/security_alerts.py b/server/secops/secops_mcp/tools/security_alerts.py index b21afa08..2a9e3cd4 100644 --- a/server/secops/secops_mcp/tools/security_alerts.py +++ b/server/secops/secops_mcp/tools/security_alerts.py @@ -24,7 +24,7 @@ # Configure logging logger = logging.getLogger('secops-mcp') -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_security_alerts( project_id: Optional[str] = None, customer_id: Optional[str] = None, @@ -153,7 +153,7 @@ async def get_security_alerts( except Exception as e: return f'Error retrieving security alerts: {str(e)}' -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_security_alert_by_id( project_id: Optional[str] = None, customer_id: Optional[str] = None, @@ -208,7 +208,12 @@ async def get_security_alert_by_id( return json.dumps(response) -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def do_update_security_alert( project_id: Optional[str] = None, customer_id: Optional[str] = None, diff --git a/server/secops/secops_mcp/tools/security_events.py b/server/secops/secops_mcp/tools/security_events.py index b436c1e7..fb41cf90 100644 --- a/server/secops/secops_mcp/tools/security_events.py +++ b/server/secops/secops_mcp/tools/security_events.py @@ -25,7 +25,7 @@ logger = logging.getLogger('secops-mcp') -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_security_events( text: str, project_id: Optional[str] = None, diff --git a/server/secops/secops_mcp/tools/security_rules.py b/server/secops/secops_mcp/tools/security_rules.py index ff55eedc..2dfb746c 100644 --- a/server/secops/secops_mcp/tools/security_rules.py +++ b/server/secops/secops_mcp/tools/security_rules.py @@ -23,7 +23,7 @@ logger = logging.getLogger("secops-mcp") -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_security_rules( project_id: Optional[str] = None, customer_id: Optional[str] = None, @@ -82,7 +82,7 @@ async def list_security_rules( return {"error": str(e), "rules": []} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_security_rules( query: str, project_id: Optional[str] = None, @@ -136,7 +136,7 @@ async def search_security_rules( return {"error": str(e), "rules": []} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_detection_rule( rule_id: str, project_id: Optional[str] = None, @@ -224,7 +224,7 @@ async def get_detection_rule( } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_rule_detections( rule_id: str, alert_state: Optional[str] = None, @@ -357,7 +357,7 @@ async def get_rule_detections( # Example of how list_errors might be defined as an MCP tool, if needed later. # This is based on the second function in the first code block provided by the user. -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_rule_errors( rule_id: str, project_id: Optional[str] = None, @@ -427,7 +427,12 @@ async def list_rule_errors( return {"error": f"Unexpected error: {str(e)}", "errors": []} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def create_rule( rule_text: str, project_id: Optional[str] = None, @@ -540,7 +545,7 @@ async def create_rule( return f"Error creating rule: {str(e)}" -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def test_rule( rule_text: str, project_id: Optional[str] = None, @@ -734,7 +739,7 @@ async def test_rule( return f"Error testing rule: {str(e)}" -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def validate_rule( rule_text: str, project_id: Optional[str] = None, @@ -884,7 +889,12 @@ async def validate_rule( return f"Error validating rule: {str(e)}" -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def create_retrohunt( rule_id: str, start_time: str, @@ -1044,7 +1054,7 @@ async def create_retrohunt( } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_retrohunt( rule_id: str, operation_id: str, @@ -1170,7 +1180,7 @@ async def get_retrohunt( } -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def search_rule_alerts( start_time: str, end_time: str, diff --git a/server/secops/secops_mcp/tools/threat_intel.py b/server/secops/secops_mcp/tools/threat_intel.py index 8a47fe4a..64a3a13a 100644 --- a/server/secops/secops_mcp/tools/threat_intel.py +++ b/server/secops/secops_mcp/tools/threat_intel.py @@ -23,7 +23,7 @@ # Configure logging logger = logging.getLogger('secops-mcp') -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_threat_intel( query: str, project_id: Optional[str] = None, diff --git a/server/secops/secops_mcp/tools/udm_search.py b/server/secops/secops_mcp/tools/udm_search.py index 3ceeb737..ef8d5c13 100644 --- a/server/secops/secops_mcp/tools/udm_search.py +++ b/server/secops/secops_mcp/tools/udm_search.py @@ -25,7 +25,7 @@ logger = logging.getLogger("secops-mcp") -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def export_udm_search_csv( query: str, fields: List[str], @@ -199,7 +199,7 @@ async def export_udm_search_csv( return f"Error exporting UDM search results: {str(e)}" -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def find_udm_field_values( query: str, page_size: Optional[int] = None, diff --git a/server/secops/secops_mcp/tools/watchlist_management.py b/server/secops/secops_mcp/tools/watchlist_management.py index 5e4ee821..21dbbd44 100644 --- a/server/secops/secops_mcp/tools/watchlist_management.py +++ b/server/secops/secops_mcp/tools/watchlist_management.py @@ -22,7 +22,12 @@ logger = logging.getLogger("secops-mcp") -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def create_watchlist( name: str, display_name: str, @@ -122,7 +127,12 @@ async def create_watchlist( return {"error": f"Error creating watchlist {name}: {str(e)}"} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": False, + } +) async def update_watchlist( watchlist_id: str, display_name: Optional[str] = None, @@ -223,7 +233,12 @@ async def update_watchlist( return {"error": f"Error updating watchlist {watchlist_id}: {str(e)}"} -@server.tool() +@server.tool( + annotations={ + "readOnlyHint": False, + "destructiveHint": True, + } +) async def delete_watchlist( watchlist_id: str, force: bool = False, @@ -318,7 +333,7 @@ async def delete_watchlist( return {"error": f"Error deleting watchlist {watchlist_id}: {str(e)}"} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def get_watchlist( watchlist_id: str, project_id: Optional[str] = None, @@ -399,7 +414,7 @@ async def get_watchlist( return {"error": error_msg} -@server.tool() +@server.tool(annotations={"readOnlyHint": True}) async def list_watchlists( page_size: Optional[int] = None, page_token: Optional[str] = None, diff --git a/server/secops/tests/test_tool_annotations_unit.py b/server/secops/tests/test_tool_annotations_unit.py new file mode 100644 index 00000000..669d94cc --- /dev/null +++ b/server/secops/tests/test_tool_annotations_unit.py @@ -0,0 +1,168 @@ +# Copyright 2025 Google LLC +# +# 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. +"""Unit tests for FastMCP tool annotations in secops-mcp.""" + +import pytest +from mcp.types import ToolAnnotations + +import secops_mcp.server as secops_server + +DESTRUCTIVE_TOOLS = [ + "delete_data_table_rows", + "delete_feed", + "delete_watchlist", + "disable_feed", + "generate_feed_secret", + "deactivate_parser", +] + +MUTATING_ADDITIVE_TOOLS = [ + "activate_parser", + "add_rows_to_data_table", + "create_data_table", + "create_feed", + "create_parser", + "create_reference_list", + "create_retrohunt", + "create_rule", + "create_rule_exclusion", + "create_watchlist", + "do_update_security_alert", + "enable_feed", + "ingest_raw_log", + "ingest_udm_events", + "patch_rule_exclusion", + "trigger_investigation", + "update_curated_rule_set_deployment", + "update_feed", + "update_reference_list", + "update_rule_exclusion_deployment", + "update_watchlist", +] + +READ_ONLY_TOOLS = [ + "compute_rule_exclusion_activity", + "export_udm_search_csv", + "fetch_associated_investigations", + "find_udm_field_values", + "get_available_log_types", + "get_curated_rule", + "get_curated_rule_by_name", + "get_curated_rule_set", + "get_detection_rule", + "get_feed", + "get_investigation", + "get_ioc_matches", + "get_parser", + "get_reference_list", + "get_retrohunt", + "get_rule_detections", + "get_rule_exclusion", + "get_security_alert_by_id", + "get_security_alerts", + "get_threat_intel", + "get_watchlist", + "list_curated_rule_set_deployments", + "list_curated_rule_sets", + "list_curated_rules", + "list_data_table_rows", + "list_feeds", + "list_investigations", + "list_parsers", + "list_rule_errors", + "list_rule_exclusions", + "list_security_rules", + "list_watchlists", + "lookup_entity", + "run_parser_against_sample_logs", + "search_curated_detections", + "search_rule_alerts", + "search_security_events", + "search_security_rules", + "search_udm", + "test_rule", + "validate_rule", +] + + +def test_tool_catalog_coverage(): + """Verify our test suites cover all 68 tools registered on the server.""" + tools = secops_server.server._tool_manager.list_tools() + all_registered_names = {t.name for t in tools} + tested_names = set(DESTRUCTIVE_TOOLS) | set(MUTATING_ADDITIVE_TOOLS) | set(READ_ONLY_TOOLS) + + assert all_registered_names == tested_names, ( + f"Mismatch in tested tools vs registered tools. " + f"Missing from tests: {all_registered_names - tested_names}. " + f"Extra in tests: {tested_names - all_registered_names}" + ) + + +def test_all_tools_have_annotations(): + """Verify every registered tool defines explicit ToolAnnotations with readOnlyHint.""" + tools = secops_server.server._tool_manager.list_tools() + assert len(tools) > 0, "No tools registered on server" + + missing_annotations = [] + missing_hints = [] + + for tool in tools: + if tool.annotations is None: + missing_annotations.append(tool.name) + continue + + assert isinstance(tool.annotations, ToolAnnotations) + if tool.annotations.readOnlyHint is None: + missing_hints.append(f"{tool.name}: readOnlyHint is None") + + if tool.annotations.readOnlyHint is False and tool.annotations.destructiveHint is None: + missing_hints.append(f"{tool.name}: destructiveHint is None for mutating tool") + + assert not missing_annotations, ( + f"The following {len(missing_annotations)} tools are missing annotations: " + f"{', '.join(sorted(missing_annotations))}" + ) + assert not missing_hints, ( + f"The following tools have incomplete hint definitions: " + f"{', '.join(sorted(missing_hints))}" + ) + + +@pytest.mark.parametrize("tool_name", DESTRUCTIVE_TOOLS) +def test_destructive_tool_annotations(tool_name: str): + """Verify destructive tools declare readOnlyHint=False and destructiveHint=True.""" + tool = secops_server.server._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is False, f"Expected {tool_name} readOnlyHint=False" + assert tool.annotations.destructiveHint is True, f"Expected {tool_name} destructiveHint=True" + + +@pytest.mark.parametrize("tool_name", MUTATING_ADDITIVE_TOOLS) +def test_mutating_additive_tool_annotations(tool_name: str): + """Verify additive mutating tools declare readOnlyHint=False and destructiveHint=False.""" + tool = secops_server.server._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is False, f"Expected {tool_name} readOnlyHint=False" + assert tool.annotations.destructiveHint is False, f"Expected {tool_name} destructiveHint=False" + + +@pytest.mark.parametrize("tool_name", READ_ONLY_TOOLS) +def test_read_only_tool_annotations(tool_name: str): + """Verify query and search tools declare readOnlyHint=True.""" + tool = secops_server.server._tool_manager.get_tool(tool_name) + assert tool is not None, f"Tool {tool_name} not found" + assert tool.annotations is not None, f"Tool {tool_name} has no annotations" + assert tool.annotations.readOnlyHint is True, f"Expected {tool_name} readOnlyHint=True"