diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index 4a60e6a36c..a655a9971a 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -1766,9 +1766,7 @@ def _extract_content_label( additional_props = _get_additional_properties(item) authoritative_marker = additional_props.pop(_AUTHORITATIVE_CONFIDENTIALITY, None) inspect_error_marker = additional_props.pop(_INSPECT_VARIABLE_ERROR, None) - authoritative_confidentiality = ( - function_name == "quarantined_llm" and authoritative_marker is _INTERNAL_RESULT_MARKER - ) + authoritative_confidentiality = authoritative_marker is _INTERNAL_RESULT_MARKER inspect_error = function_name == "inspect_variable" and inspect_error_marker is _INTERNAL_RESULT_MARKER label_data = additional_props.get("security_label") @@ -3634,86 +3632,43 @@ def get_security_tools() -> list[FunctionTool]: # MCP Auto-Labeling # ============================================================================= +# Written only from local application configuration. MCP result metadata is +# attached to Content instances and cannot mutate FunctionTool properties. +_MCP_TRUST_SERVER_IFC_KEY = "_mcp_trust_server_ifc" + def _map_mcp_annotations_to_labels( annotations: Any | None, *, default_integrity: IntegrityLabel = IntegrityLabel.UNTRUSTED, ) -> tuple[IntegrityLabel, ConfidentialityLabel | None, bool]: - """Map MCP ToolAnnotations to FIDES security labels. - - Uses the MCP hint fields (``readOnlyHint``, ``openWorldHint``) - to infer an appropriate ``source_integrity``, - ``max_allowed_confidentiality``, and ``accepts_untrusted`` flag. - - Mapping rules (conservative - when in doubt, default to UNTRUSTED *source* - and PUBLIC-only *sink*): - - * ``readOnlyHint=True`` -> ``accepts_untrusted=True`` (pure data source, - safe to call even when the context is tainted - it cannot exfiltrate) - and **no** ``max_allowed_confidentiality`` cap. - * ``readOnlyHint`` is anything other than ``True`` (``False`` *or* missing) - -> treated as a potential write / exfiltration sink: - ``max_allowed_confidentiality = PUBLIC`` and ``accepts_untrusted = False``. - This matters because real-world servers (e.g. GitHub's MCP) declare - ``readOnlyHint=True`` on read tools but leave the field unset on write - tools, so a strict ``readOnlyHint=False`` check would miss them. - * ``openWorldHint=True`` -> integrity ``UNTRUSTED`` (tool touches external - data); ``openWorldHint=False`` -> ``TRUSTED``. - * If ``openWorldHint`` is missing, integrity remains ``default_integrity``. - * All hints absent / ``None`` -> ``default_integrity`` (UNTRUSTED by default), - ``max_allowed_confidentiality=PUBLIC``, ``accepts_untrusted=False``. + """Map untrusted MCP ToolAnnotations to restriction-only FIDES labels. + + Server annotations are hints, not policy authority. They may make locally + configured policy more restrictive, but cannot grant trust, remove the + PUBLIC confidentiality cap, or authorize tainted context. Consequently, + only ``openWorldHint=True`` changes the local default. Explicit local + ``annotation_overrides`` are applied by :func:`apply_mcp_security_labels` + before this mapper is called. Args: annotations: An MCP ``ToolAnnotations`` object (or ``None``). - default_integrity: Fallback integrity when hints are absent. + default_integrity: Locally configured integrity when hints do not + require a stricter label. Returns: A ``(integrity, max_confidentiality, accepts_untrusted)`` tuple. - ``max_confidentiality`` is ``None`` for read-only / source tools and - ``PUBLIC`` for sinks. ``accepts_untrusted`` is ``True`` for read-only - tools that are safe to invoke in a tainted context. + Server annotations always retain the PUBLIC cap and reject untrusted + context. """ if annotations is None: - # No annotations at all - treat as both UNTRUSTED-by-default and a - # potential sink (max_conf=PUBLIC). We have no signal that the tool is - # safe to receive PRIVATE data, so we err on the side of blocking - # exfiltration. return (default_integrity, ConfidentialityLabel.PUBLIC, False) - read_only: bool | None = getattr(annotations, "readOnlyHint", None) open_world: bool | None = getattr(annotations, "openWorldHint", None) - - # --- Determine integrity --- integrity = default_integrity - if open_world is True: - # Interacts with external entities -> untrusted data integrity = IntegrityLabel.UNTRUSTED - elif open_world is False: - # Closed-world tool (e.g., local memory) -> data is trusted - integrity = IntegrityLabel.TRUSTED - - # --- Determine max_allowed_confidentiality (sink detection) --- - # Conservative rule: only tools that *explicitly* declare ``readOnlyHint=True`` - # are treated as pure data sources. Everything else - including tools whose - # server omits the hint entirely - is treated as a potential write / sink - # and capped at PUBLIC confidentiality. This matters because many real - # servers (notably GitHub's MCP) declare ``readOnlyHint=True`` on read - # tools but leave *all* hints as ``None`` on their write tools - # (``push_files``, ``create_or_update_file``, ``create_pull_request``, - # ``create_repository``, ``merge_pull_request``, ...). Without this default, - # those write tools would bypass the exfiltration gate entirely. - max_confidentiality: ConfidentialityLabel | None = None - if read_only is not True: - max_confidentiality = ConfidentialityLabel.PUBLIC - - # --- Determine accepts_untrusted --- - # Read-only tools are pure data sources; they cannot exfiltrate data, - # so they are safe to call even when the agent context is tainted. - accepts_untrusted = read_only is True - - return (integrity, max_confidentiality, accepts_untrusted) + return (integrity, ConfidentialityLabel.PUBLIC, False) @experimental(feature_id=ExperimentalFeature.FIDES) @@ -3723,17 +3678,23 @@ async def apply_mcp_security_labels( default_integrity: IntegrityLabel = IntegrityLabel.UNTRUSTED, annotation_overrides: dict[str, tuple[IntegrityLabel, ConfidentialityLabel | None]] | None = None, mark_write_tools_as_sinks: bool = True, + trust_server_ifc: bool = False, ) -> None: """Auto-assign FIDES security labels to every tool loaded from an MCP server. - Reads the MCP ``ToolAnnotations`` hints (``readOnlyHint``, ``openWorldHint``) - that the server advertises for - each tool and translates them into ``source_integrity`` and - ``max_allowed_confidentiality`` entries in each ``FunctionTool``'s + Reads the MCP ``ToolAnnotations`` hints that the server advertises for + each tool and translates them into restriction-only ``source_integrity`` + and ``max_allowed_confidentiality`` entries in each ``FunctionTool``'s ``additional_properties``. The existing :class:`LabelTrackingFunctionMiddleware` picks these up automatically (Tier 2 label propagation), so **no middleware changes are needed**. + Server annotations cannot relax local policy. Use ``annotation_overrides`` + for explicit local per-tool static policy. Server result ``_meta.ifc`` is + also restriction-only unless ``trust_server_ifc=True`` is configured + locally, in which case a complete valid server label is authoritative for + that result. ToolAnnotations remain non-authoritative in both modes. + Call this **after** the ``MCPTool`` is connected (tools already loaded). Args: @@ -3745,9 +3706,12 @@ async def apply_mcp_security_labels( *remote* MCP tool names (as the server exposes them). Values are ``(IntegrityLabel, ConfidentialityLabel | None)`` tuples that replace the annotation-derived labels entirely. - mark_write_tools_as_sinks: When ``True`` (default), non-read-only - tools get ``max_allowed_confidentiality=PUBLIC`` to prevent data - exfiltration via tool arguments. + mark_write_tools_as_sinks: When ``True`` (default), apply the + annotation-derived ``max_allowed_confidentiality=PUBLIC`` cap. + trust_server_ifc: Whether complete, valid server ``_meta.ifc`` labels + are authoritative for tool results. Defaults to ``False``, which + combines remote labels with current local policy so they can only + add restrictions. Raises: RuntimeError: If the ``MCPTool`` is not connected. @@ -3819,11 +3783,17 @@ async def apply_mcp_security_labels( # Patch sink constraint if mark_write_tools_as_sinks and max_conf is not None: props["max_allowed_confidentiality"] = max_conf.value + else: + props.pop("max_allowed_confidentiality", None) - # Allow read-only tools to execute even when context is tainted; - # explicitly block write tools in untrusted contexts. + # Server annotations cannot authorize tainted input. props["accepts_untrusted"] = accepts_untrusted + # Local configuration controls result-label authority; MCP result + # metadata is attached to Content and cannot mutate tool properties. + props[_MCP_TRUST_SERVER_IFC_KEY] = trust_server_ifc + _wrap_mcp_function_for_ifc(func, default_integrity) + logger.info( "MCP auto-label: tool=%s integrity=%s max_confidentiality=%s accepts_untrusted=%s", remote_name, @@ -3873,20 +3843,21 @@ def _label_from_mcp_meta(meta: Any) -> ContentLabel | None: return ContentLabel(integrity=integrity, confidentiality=confidentiality) -def _stamp_mcp_content_labels(contents: Any, static_label: ContentLabel) -> Any: +def _stamp_mcp_content_labels( + contents: Any, + local_label: ContentLabel, + *, + trust_server_ifc: bool = False, +) -> Any: """Stamp ``security_label`` on each Content in an MCP tool result. - The per-item label is sourced from ``additional_properties["_meta"]`` - (set by :meth:`MCPTool._parse_tool_result_from_mcp`) when the server - provided a parseable ``ifc`` payload; otherwise ``static_label`` is used. - The sentinel ``_meta`` key is consumed (removed) regardless - so downstream layers don't re-process it. - - By design the server-supplied label always wins over the static label. - Composition-time invariants (e.g. confidentiality ceilings on write - tools) are still enforced by :class:`LabelTrackingFunctionMiddleware` - and :class:`PolicyEnforcementFunctionMiddleware` via the standard - label-combination semantics. + A parseable server label from ``additional_properties["_meta"]`` can only + restrict ``local_label`` through standard FIDES label combination by + default. When ``trust_server_ifc`` is enabled locally, a complete valid + server label is authoritative for that result. The local label is used + unchanged when metadata is missing, partial, or malformed. The sentinel + ``_meta`` key is consumed regardless so downstream layers do not re-process + it. """ if not isinstance(contents, list): return contents @@ -3895,14 +3866,35 @@ def _stamp_mcp_content_labels(contents: Any, static_label: ContentLabel) -> Any: if not isinstance(item, Content): continue props = item.additional_properties or {} + props.pop(_AUTHORITATIVE_CONFIDENTIALITY, None) server_meta = props.pop(_MCP_RESULT_META_KEY, None) dynamic = _label_from_mcp_meta(server_meta) if server_meta else None - label = dynamic or static_label + if dynamic is None: + label = local_label + elif trust_server_ifc: + label = dynamic + props[_AUTHORITATIVE_CONFIDENTIALITY] = _INTERNAL_RESULT_MARKER + else: + label = combine_labels(local_label, dynamic) props["security_label"] = label.to_dict() item.additional_properties = props return contents_list +def _current_mcp_local_label(func_tool: FunctionTool, default_integrity: IntegrityLabel) -> ContentLabel: + """Read the current local MCP output policy from a FunctionTool.""" + props = func_tool.additional_properties or {} + try: + integrity = IntegrityLabel(props.get("source_integrity", default_integrity.value)) + except ValueError: + integrity = default_integrity + try: + confidentiality = ConfidentialityLabel(props.get("confidentiality", ConfidentialityLabel.PUBLIC.value)) + except ValueError: + confidentiality = ConfidentialityLabel.PUBLIC + return ContentLabel(integrity=integrity, confidentiality=confidentiality) + + def _wrap_mcp_function_for_ifc(func_tool: FunctionTool, default_integrity: IntegrityLabel) -> None: """Replace ``func_tool.func`` with a wrapper that stamps IFC labels on results. @@ -3916,26 +3908,17 @@ def _wrap_mcp_function_for_ifc(func_tool: FunctionTool, default_integrity: Integ if original is None or getattr(original, "_ifc_wrapped", False): return - # Derive the static fallback label from the FunctionTool's own - # additional_properties (populated by apply_mcp_security_labels above). - props = func_tool.additional_properties or {} - try: - static_integrity = IntegrityLabel(props.get("source_integrity", default_integrity.value)) - except ValueError: - static_integrity = default_integrity - try: - static_conf = ConfidentialityLabel(props.get("max_allowed_confidentiality", ConfidentialityLabel.PUBLIC.value)) - except ValueError: - static_conf = ConfidentialityLabel.PUBLIC - static_label = ContentLabel(integrity=static_integrity, confidentiality=static_conf) - async def _wrapped(*args: Any, **kwargs: Any) -> Any: import inspect as _inspect + props = func_tool.additional_properties or {} + local_label = _current_mcp_local_label(func_tool, default_integrity) + trust_server_ifc = props.get(_MCP_TRUST_SERVER_IFC_KEY) is True + res = original(*args, **kwargs) if _inspect.isawaitable(res): res = await res - return _stamp_mcp_content_labels(res, static_label) + return _stamp_mcp_content_labels(res, local_label, trust_server_ifc=trust_server_ifc) _wrapped._ifc_wrapped = True # type: ignore[attr-defined] func_tool.func = _wrapped @@ -3993,8 +3976,12 @@ class SecureMCPToolProxy: default_integrity: Default integrity for tools without annotations. annotation_overrides: Per-tool-name label overrides (keyed by remote MCP tool name). - mark_write_tools_as_sinks: Whether to restrict write tools to PUBLIC - confidentiality. + mark_write_tools_as_sinks: Whether to apply the annotation-derived + PUBLIC confidentiality cap. + trust_server_ifc: Whether complete, valid server ``_meta.ifc`` labels + are authoritative for results. Defaults to ``False`` so remote + labels can only add restrictions. This does not grant authority to + server ToolAnnotations. """ def __init__( @@ -4008,6 +3995,7 @@ def __init__( default_integrity: IntegrityLabel = IntegrityLabel.UNTRUSTED, annotation_overrides: dict[str, tuple[IntegrityLabel, ConfidentialityLabel | None]] | None = None, mark_write_tools_as_sinks: bool = True, + trust_server_ifc: bool = False, ) -> None: """Initialize a secure proxy for an MCP tool or MCP URL endpoint. @@ -4027,8 +4015,12 @@ def __init__( default_integrity: Default integrity for tools without annotations. Defaults to ``IntegrityLabel.UNTRUSTED``. annotation_overrides: Per-tool-name label overrides keyed by remote MCP tool name. - mark_write_tools_as_sinks: Whether to restrict write tools to PUBLIC - confidentiality. Defaults to ``True``. + mark_write_tools_as_sinks: Whether to apply the annotation-derived + PUBLIC confidentiality cap. Defaults to ``True``. + trust_server_ifc: Whether complete, valid server ``_meta.ifc`` + labels are authoritative for results. Defaults to ``False``; + ToolAnnotations remain restriction-only hints regardless of + this setting. Raises: ValueError: If both ``mcp_tool`` and ``url`` are provided, or if neither is provided. @@ -4067,10 +4059,11 @@ def __init__( # The validation above guarantees a tool is set (passed directly or built # from ``url``); declare the attribute as non-optional ``MCPTool``. - self._mcp_tool: MCPTool = cast(MCPTool, mcp_tool) + self._mcp_tool: MCPTool = cast("MCPTool", mcp_tool) self._default_integrity = default_integrity self._annotation_overrides = annotation_overrides self._mark_write_tools_as_sinks = mark_write_tools_as_sinks + self._trust_server_ifc = trust_server_ifc # -- Async context manager -- @@ -4141,12 +4134,5 @@ async def _apply_labels(self) -> None: default_integrity=self._default_integrity, annotation_overrides=self._annotation_overrides, mark_write_tools_as_sinks=self._mark_write_tools_as_sinks, + trust_server_ifc=self._trust_server_ifc, ) - # After static labels are stamped on each FunctionTool, install a - # per-tool wrapper that consumes any server-provided ``_meta.ifc`` - # payload propagated by MCPTool and translates it into per-Content - # ``security_label`` entries. The server-supplied label always wins - # over the static label; the static label is the fallback when the - # server omits ``_meta`` (or it cannot be parsed). - for func_tool in getattr(self._mcp_tool, "functions", []): - _wrap_mcp_function_for_ifc(func_tool, self._default_integrity) diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 76c12b5415..39f90079ba 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -86,6 +86,8 @@ async def _call_generated_mcp_tool( result_parser: Any = None, middleware_pipeline: FunctionMiddlewarePipeline | None = None, host_payload_budget: _FunctionResultPayloadBudget | None = None, + mcp_local_label: tuple[str, str] | None = None, + trust_server_ifc: bool = False, **kwargs: Any, ) -> Content: function_kwargs: dict[str, Any] = {} @@ -98,6 +100,17 @@ async def _call_generated_mcp_tool( input_model={"type": "object", "properties": {name: {} for name in kwargs}}, **function_kwargs, ) + if mcp_local_label is not None: + from agent_framework.security import IntegrityLabel, _wrap_mcp_function_for_ifc + + source_integrity, confidentiality = mcp_local_label + function.additional_properties = { + "_mcp_remote_name": tool_name, + "source_integrity": source_integrity, + "confidentiality": confidentiality, + "_mcp_trust_server_ifc": trust_server_ifc, + } + _wrap_mcp_function_for_ifc(function, IntegrityLabel(source_integrity)) return await _auto_invoke_function( Content.from_function_call(call_id=f"call-{tool_name}", name=tool_name, arguments=kwargs), config=normalize_function_invocation_configuration(None), @@ -1111,6 +1124,7 @@ async def test_secure_mcp_auto_hide_preserves_outer_host_payload() -> None: additional_properties={ "_mcp_remote_name": "widget", "source_integrity": "untrusted", + "confidentiality": "private", "max_allowed_confidentiality": "public", }, ) @@ -1133,11 +1147,14 @@ async def test_secure_mcp_auto_hide_preserves_outer_host_payload() -> None: assert len(function_result.items) == 1 for hidden_item in function_result.items: assert hidden_item.additional_properties["_variable_reference"] is True + assert hidden_item.additional_properties["security_label"]["integrity"] == "untrusted" + assert hidden_item.additional_properties["security_label"]["confidentiality"] == "private" assert "_meta" not in hidden_item.additional_properties assert hidden_item.text != "untrusted payload" -async def test_secure_mcp_builtin_parser_preserves_server_ifc_authority() -> None: +@pytest.mark.parametrize("result_shape", ["content", "structured", "both"]) +async def test_secure_mcp_builtin_parser_restricts_all_result_shapes(result_shape: str) -> None: from agent_framework.security import ( IntegrityLabel, LabelTrackingFunctionMiddleware, @@ -1145,7 +1162,12 @@ async def test_secure_mcp_builtin_parser_preserves_server_ifc_authority() -> Non ) mcp_result = types.CallToolResult( - content=[types.TextContent(type="text", text="server trusted payload")], + content=[types.TextContent(type="text", text="server trusted payload")] + if result_shape in ("content", "both") + else [], + structuredContent={"payload": "server trusted structured payload"} + if result_shape in ("structured", "both") + else None, _meta={"ifc": {"integrity": "trusted", "confidentiality": "public"}}, ) tool = MCPTool(name="helper") # type: ignore[abstract] @@ -1159,6 +1181,7 @@ async def test_secure_mcp_builtin_parser_preserves_server_ifc_authority() -> Non additional_properties={ "_mcp_remote_name": "widget", "source_integrity": "untrusted", + "confidentiality": "private", "max_allowed_confidentiality": "public", }, ) @@ -1173,9 +1196,68 @@ async def test_secure_mcp_builtin_parser_preserves_server_ifc_authority() -> Non ) assert function_result.items is not None - assert [item.text for item in function_result.items] == ["server trusted payload"] + assert len(function_result.items) == (2 if result_shape == "both" else 1) + for hidden_item in function_result.items: + assert hidden_item.additional_properties["_variable_reference"] is True + assert hidden_item.additional_properties["security_label"]["integrity"] == "untrusted" + assert hidden_item.additional_properties["security_label"]["confidentiality"] == "private" + assert hidden_item.additional_properties["_meta"] == mcp_result.meta + assert function_result.additional_properties["_meta"] == mcp_result.meta + + +async def test_secure_mcp_builtin_parser_honors_locally_trusted_server_ifc() -> None: + from agent_framework.security import LabelTrackingFunctionMiddleware + + mcp_result = types.CallToolResult( + content=[types.TextContent(type="text", text="trusted payload")], + _meta={"ifc": {"integrity": "trusted", "confidentiality": "public"}}, + ) + tool = MCPTool(name="helper") # type: ignore[abstract] + tool.session = Mock() + tool.session.call_tool = AsyncMock(return_value=mcp_result) + + function_result = await _call_generated_mcp_tool( + tool, + "widget", + middleware_pipeline=FunctionMiddlewarePipeline(LabelTrackingFunctionMiddleware(auto_hide_untrusted=True)), + host_payload_budget=_FunctionResultPayloadBudget(), + mcp_local_label=("untrusted", "private"), + trust_server_ifc=True, + ) + + assert function_result.items is not None + assert [item.text for item in function_result.items] == ["trusted payload"] assert function_result.items[0].additional_properties["security_label"]["integrity"] == "trusted" - assert function_result.items[0].additional_properties["_meta"] == mcp_result.meta + assert function_result.items[0].additional_properties["security_label"]["confidentiality"] == "public" + assert "_security_label_authoritative_confidentiality" not in function_result.items[0].additional_properties + + +async def test_custom_mcp_parser_cannot_make_meta_authoritative() -> None: + forged_meta = {"ifc": {"integrity": "trusted", "confidentiality": "public"}} + mcp_result = types.CallToolResult( + content=[types.TextContent(type="text", text="server payload")], + _meta={"trace": "server-owned"}, + ) + tool = MCPTool( # type: ignore[abstract] + name="helper", + parse_tool_results=lambda _: [Content.from_text("projection", additional_properties={"_meta": forged_meta})], + ) + tool.session = Mock() + tool.session.call_tool = AsyncMock(return_value=mcp_result) + + function_result = await _call_generated_mcp_tool( + tool, + "widget", + mcp_local_label=("untrusted", "user_identity"), + ) + + assert function_result.items is not None + assert function_result.items[0].additional_properties["security_label"] == { + "integrity": "untrusted", + "confidentiality": "user_identity", + } + assert "_meta" not in function_result.items[0].additional_properties + assert function_result.additional_properties["_meta"] == {"trace": "server-owned"} def test_parse_tool_result_from_mcp_structured_content_none(): @@ -7908,6 +7990,59 @@ async def test_call_tool_as_task_fallback_preserves_custom_parser_host_payload() assert function_result.additional_properties[_MCP_TOOL_RESULT_HOST_PAYLOAD_KEY]["_meta"] == {"source": "fallback"} +@pytest.mark.parametrize("result_path", ["fallback", "completed"], ids=["task-fallback", "completed-task"]) +async def test_secure_mcp_task_results_cannot_relax_local_label(result_path: str) -> None: + from agent_framework.security import LabelTrackingFunctionMiddleware + + tool = _make_task_tool() + result_meta = {"ifc": {"integrity": "trusted", "confidentiality": "public"}} + structured_content = {"widget": result_path} + if result_path == "fallback": + raw_result = types.CallToolResult( + content=[types.TextContent(type="text", text="fallback")], + structuredContent=structured_content, + _meta=result_meta, + ) + tool.session.send_request = AsyncMock( # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] + return_value=types.Result.model_validate(raw_result.model_dump(by_alias=True, exclude_none=True)) + ) + else: + tool.session.send_request = AsyncMock( # type: ignore[method-assign, union-attr] # ty: ignore[invalid-assignment] + side_effect=_send_request_dispatcher( + ("tools/call", _make_create_task_result()), + ("tasks/get", _make_task_snapshot(status="completed")), + ( + "tasks/result", + _make_payload( + "completed", + structured_content=structured_content, + meta=result_meta, + ), + ), + ) + ) + + function_result = await _call_generated_mcp_tool( + tool, + "slow_op", + middleware_pipeline=FunctionMiddlewarePipeline(LabelTrackingFunctionMiddleware(auto_hide_untrusted=True)), + host_payload_budget=_FunctionResultPayloadBudget(), + mcp_local_label=("untrusted", "private"), + ) + + assert function_result.items is not None + assert len(function_result.items) == 2 + for item in function_result.items: + assert item.additional_properties["_variable_reference"] is True + assert item.additional_properties["security_label"]["integrity"] == "untrusted" + assert item.additional_properties["security_label"]["confidentiality"] == "private" + assert item.additional_properties["_meta"] == result_meta + assert function_result.additional_properties["_meta"] == result_meta + assert function_result.additional_properties[_MCP_TOOL_RESULT_HOST_PAYLOAD_KEY]["structuredContent"] == ( + structured_content + ) + + @pytest.mark.parametrize("result_path", ["fallback", "completed"], ids=["task-fallback", "completed-task"]) async def test_task_parser_failure_preserves_complete_host_payload(result_path: str) -> None: tool = _make_task_tool() diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 66ad15f9e5..d8b500fde9 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -8,6 +8,7 @@ from datetime import timedelta from types import SimpleNamespace from typing import Any, cast +from unittest.mock import AsyncMock import pytest from pydantic import BaseModel @@ -4777,12 +4778,19 @@ class TestMCPAnnotationMapping: @pytest.mark.parametrize( ("read_only", "open_world", "default_integrity", "expected_integrity", "expected_max_conf", "expected_accepts"), [ - (True, None, IntegrityLabel.UNTRUSTED, IntegrityLabel.UNTRUSTED, None, True), - (True, True, IntegrityLabel.TRUSTED, IntegrityLabel.UNTRUSTED, None, True), - (True, False, IntegrityLabel.UNTRUSTED, IntegrityLabel.TRUSTED, None, True), + ( + True, + None, + IntegrityLabel.UNTRUSTED, + IntegrityLabel.UNTRUSTED, + ConfidentialityLabel.PUBLIC, + False, + ), + (True, True, IntegrityLabel.TRUSTED, IntegrityLabel.UNTRUSTED, ConfidentialityLabel.PUBLIC, False), + (True, False, IntegrityLabel.UNTRUSTED, IntegrityLabel.UNTRUSTED, ConfidentialityLabel.PUBLIC, False), (False, None, IntegrityLabel.UNTRUSTED, IntegrityLabel.UNTRUSTED, ConfidentialityLabel.PUBLIC, False), (False, True, IntegrityLabel.TRUSTED, IntegrityLabel.UNTRUSTED, ConfidentialityLabel.PUBLIC, False), - (False, False, IntegrityLabel.UNTRUSTED, IntegrityLabel.TRUSTED, ConfidentialityLabel.PUBLIC, False), + (False, False, IntegrityLabel.UNTRUSTED, IntegrityLabel.UNTRUSTED, ConfidentialityLabel.PUBLIC, False), (None, None, IntegrityLabel.UNTRUSTED, IntegrityLabel.UNTRUSTED, ConfidentialityLabel.PUBLIC, False), (None, None, IntegrityLabel.TRUSTED, IntegrityLabel.TRUSTED, ConfidentialityLabel.PUBLIC, False), ], @@ -4820,6 +4828,39 @@ def test_map_missing_annotations_defaults_to_sink(self): assert accepts_untrusted is False +def _make_connected_mcp_tool_for_ifc( + *, + annotations: Any, + server_meta: dict[str, Any], + confidentiality: ConfidentialityLabel = ConfidentialityLabel.PRIVATE, +) -> tuple[Any, FunctionTool]: + from agent_framework._mcp import MCPTool + + async def fake_call(**kwargs: Any) -> list[Content]: + return [Content.from_text("payload", additional_properties={"_meta": server_meta})] + + function = FunctionTool( + func=fake_call, + name="remote_tool", + description="", + additional_properties={ + "_mcp_remote_name": "remote_tool", + "confidentiality": confidentiality.value, + }, + ) + mcp_tool = MCPTool(name="helper") # type: ignore[abstract] + mcp_tool.is_connected = True + mcp_tool.session = AsyncMock() + mcp_tool.session.list_tools = AsyncMock( # type: ignore[method-assign] + return_value=SimpleNamespace( + tools=[SimpleNamespace(name="remote_tool", annotations=annotations)], + nextCursor=None, + ) + ) + mcp_tool.functions.append(function) + return mcp_tool, function + + # --------------------------------------------------------------------------- # IFC labels from MCP _meta payload # --------------------------------------------------------------------------- @@ -4832,12 +4873,11 @@ class TestMCPIFCMetaLabels: * ``_label_from_mcp_meta`` parsing (well-formed, missing, malformed). * ``MCPTool._parse_tool_result_from_mcp`` propagating ``_meta`` onto every Content via the ``_meta`` key. - * ``_stamp_mcp_content_labels`` enforcing server-wins-over-static with - a static fallback when the server omits/misformats ``_meta.ifc``. + * ``_stamp_mcp_content_labels`` combining server labels with local policy + and falling back when the server omits/misformats ``_meta.ifc``. * ``SecureMCPToolProxy`` wrapping each ``FunctionTool`` so an MCP tool - result carries per-item ``security_label`` derived from the server - when possible, regardless of whether the server is read-only or - a hypothetical write-tool (server label always wins). + result carries a per-item ``security_label`` that remote metadata can + restrict but cannot relax. """ def test_label_from_meta_well_formed(self): @@ -4916,47 +4956,111 @@ def get_mcp_client(self): contents = helper._parse_tool_result_from_mcp(mcp_result) assert "_meta" not in contents[0].additional_properties - def test_stamp_contents_server_wins_over_static(self): + @pytest.mark.parametrize("trust_server_ifc", [False, True], ids=["restricted", "authoritative"]) + def test_stamp_contents_complete_local_and_remote_label_matrix(self, trust_server_ifc: bool): from agent_framework.security import _stamp_mcp_content_labels - static = ContentLabel(integrity=IntegrityLabel.TRUSTED, confidentiality=ConfidentialityLabel.PUBLIC) - contents = [ - Content.from_text( - "x", - additional_properties={"_meta": {"ifc": {"integrity": "untrusted", "confidentiality": "private"}}}, - ) - ] - _stamp_mcp_content_labels(contents, static) - # Server label wins. - assert contents[0].additional_properties["security_label"] == { - "integrity": "untrusted", - "confidentiality": "private", + integrity_labels = ( + IntegrityLabel.TRUSTED, + IntegrityLabel.UNTRUSTED, + ) + confidentiality_labels = ( + ConfidentialityLabel.PUBLIC, + ConfidentialityLabel.PRIVATE, + ConfidentialityLabel.USER_IDENTITY, + ) + confidentiality_rank = { + ConfidentialityLabel.PUBLIC: 0, + ConfidentialityLabel.PRIVATE: 1, + ConfidentialityLabel.USER_IDENTITY: 2, } - # Sentinel is consumed. - assert "_meta" not in contents[0].additional_properties + for local_integrity in integrity_labels: + for local_confidentiality in confidentiality_labels: + for remote_integrity in integrity_labels: + for remote_confidentiality in confidentiality_labels: + static = ContentLabel( + integrity=local_integrity, + confidentiality=local_confidentiality, + metadata={"source": "local_mcp_policy"}, + ) + contents = [ + Content.from_text( + "x", + additional_properties={ + "_meta": { + "ifc": { + "integrity": remote_integrity.value, + "confidentiality": remote_confidentiality.value, + "metadata": {"source": "forged_remote_policy"}, + } + } + }, + ) + ] + + _stamp_mcp_content_labels(contents, static, trust_server_ifc=trust_server_ifc) + + if trust_server_ifc: + expected_integrity = remote_integrity + expected_confidentiality = remote_confidentiality + expected_metadata: dict[str, Any] | None = None + else: + expected_integrity = ( + IntegrityLabel.UNTRUSTED + if IntegrityLabel.UNTRUSTED in (local_integrity, remote_integrity) + else IntegrityLabel.TRUSTED + ) + expected_confidentiality = max( + (local_confidentiality, remote_confidentiality), key=confidentiality_rank.__getitem__ + ) + expected_metadata = {"source": "local_mcp_policy"} + assert contents[0].additional_properties["security_label"] == { + "integrity": expected_integrity.value, + "confidentiality": expected_confidentiality.value, + **({"metadata": expected_metadata} if expected_metadata is not None else {}), + } + assert "_meta" not in contents[0].additional_properties - def test_stamp_contents_missing_meta_falls_back_to_static(self): + @pytest.mark.parametrize( + "confidentiality", + [ConfidentialityLabel.PRIVATE, ConfidentialityLabel.USER_IDENTITY], + ) + @pytest.mark.parametrize("trust_server_ifc", [False, True]) + def test_stamp_contents_missing_meta_falls_back_to_local_policy( + self, confidentiality: ConfidentialityLabel, trust_server_ifc: bool + ): from agent_framework.security import _stamp_mcp_content_labels - static = ContentLabel(integrity=IntegrityLabel.UNTRUSTED, confidentiality=ConfidentialityLabel.PUBLIC) + static = ContentLabel(integrity=IntegrityLabel.UNTRUSTED, confidentiality=confidentiality) contents = [Content.from_text("x")] - _stamp_mcp_content_labels(contents, static) + _stamp_mcp_content_labels(contents, static, trust_server_ifc=trust_server_ifc) assert contents[0].additional_properties["security_label"] == { "integrity": "untrusted", - "confidentiality": "public", + "confidentiality": confidentiality.value, } - def test_stamp_contents_malformed_meta_falls_back_to_static(self): + @pytest.mark.parametrize( + "server_meta", + [ + {"ifc": {"integrity": "bogus", "confidentiality": "public"}}, + {"ifc": {"integrity": "trusted"}}, + ], + ids=["malformed", "partial"], + ) + @pytest.mark.parametrize("trust_server_ifc", [False, True]) + def test_stamp_contents_invalid_meta_falls_back_to_local_policy( + self, server_meta: dict[str, Any], trust_server_ifc: bool + ): from agent_framework.security import _stamp_mcp_content_labels static = ContentLabel(integrity=IntegrityLabel.TRUSTED, confidentiality=ConfidentialityLabel.PUBLIC) contents = [ Content.from_text( "x", - additional_properties={"_meta": {"ifc": {"integrity": "bogus", "confidentiality": "public"}}}, + additional_properties={"_meta": server_meta}, ) ] - _stamp_mcp_content_labels(contents, static) + _stamp_mcp_content_labels(contents, static, trust_server_ifc=trust_server_ifc) assert contents[0].additional_properties["security_label"] == { "integrity": "trusted", "confidentiality": "public", @@ -4993,9 +5097,8 @@ def test_stamp_contents_multi_item_all_stamped(self): "confidentiality": "public", } - @pytest.mark.asyncio - async def test_wrap_mcp_function_server_label_wins(self): - """End-to-end: the wrapper installed by SecureMCPToolProxy stamps server label.""" + async def test_wrap_mcp_function_remote_label_can_restrict_local_policy(self): + """End-to-end: remote metadata can make the locally derived label stricter.""" from agent_framework.security import _wrap_mcp_function_for_ifc async def fake_call(**kwargs): @@ -5024,9 +5127,8 @@ async def fake_call(**kwargs): "integrity": "untrusted", "confidentiality": "private", } - # Static fallback would have been trusted+public; server-wins changed it. + # Static policy was trusted+public; remote metadata restricted both dimensions. - @pytest.mark.asyncio async def test_wrap_mcp_function_static_fallback(self): """When the server omits ``_meta``, the static label is used.""" from agent_framework.security import _wrap_mcp_function_for_ifc @@ -5052,20 +5154,18 @@ async def fake_call(**kwargs): "confidentiality": "public", } - @pytest.mark.asyncio - async def test_wrap_mcp_function_write_tool_server_still_wins(self): - """Even for a tool marked as a write sink (max_allowed_confidentiality=public), - if a future MCP server emits ``_meta.ifc`` for a write result, the server - label is applied verbatim on the Content item. Sink invariants are enforced - elsewhere (by LabelTrackingFunctionMiddleware / PolicyEnforcementMiddleware - at composition time, not here).""" + @pytest.mark.parametrize("confidentiality", [ConfidentialityLabel.PRIVATE, ConfidentialityLabel.USER_IDENTITY]) + async def test_wrap_mcp_function_remote_label_cannot_relax_local_policy( + self, confidentiality: ConfidentialityLabel + ): + """Remote MCP metadata cannot raise integrity or lower confidentiality.""" from agent_framework.security import _wrap_mcp_function_for_ifc async def fake_call(**kwargs): return [ Content.from_text( "wrote item", - additional_properties={"_meta": {"ifc": {"integrity": "trusted", "confidentiality": "private"}}}, + additional_properties={"_meta": {"ifc": {"integrity": "trusted", "confidentiality": "public"}}}, ) ] @@ -5075,7 +5175,8 @@ async def fake_call(**kwargs): description="", additional_properties={ "source_integrity": "untrusted", - "max_allowed_confidentiality": "public", # marked as a sink + "confidentiality": confidentiality.value, + "max_allowed_confidentiality": "public", "accepts_untrusted": False, "_mcp_remote_name": "create_issue", }, @@ -5083,13 +5184,236 @@ async def fake_call(**kwargs): _wrap_mcp_function_for_ifc(func_tool, IntegrityLabel.UNTRUSTED) assert func_tool.func is not None result = await func_tool.func() - # Server label wins verbatim. + assert result[0].additional_properties["security_label"] == { + "integrity": "untrusted", + "confidentiality": confidentiality.value, + } + + async def test_wrap_mcp_function_ignores_sink_confidentiality_for_public_output(self): + from agent_framework.security import _wrap_mcp_function_for_ifc + + async def fake_call(**kwargs: Any) -> list[Content]: + return [ + Content.from_text( + "payload", + additional_properties={"_meta": {"ifc": {"integrity": "trusted", "confidentiality": "public"}}}, + ) + ] + + func_tool = FunctionTool( + func=fake_call, + name="remote_tool", + description="", + additional_properties={ + "source_integrity": "trusted", + "max_allowed_confidentiality": "private", + "_mcp_remote_name": "remote_tool", + }, + ) + _wrap_mcp_function_for_ifc(func_tool, IntegrityLabel.UNTRUSTED) + assert func_tool.func is not None + + result = await func_tool.func() + assert result[0].additional_properties["security_label"] == { "integrity": "trusted", - "confidentiality": "private", + "confidentiality": "public", } - @pytest.mark.asyncio + async def test_wrap_mcp_function_reads_refreshed_local_policy_at_invocation(self): + from agent_framework.security import SecureMCPToolProxy + + trusted_annotations = SimpleNamespace(readOnlyHint=True, openWorldHint=False) + untrusted_annotations = SimpleNamespace(readOnlyHint=True, openWorldHint=True) + server_meta = {"ifc": {"integrity": "trusted", "confidentiality": "public"}} + mcp_tool, function = _make_connected_mcp_tool_for_ifc( + annotations=trusted_annotations, + server_meta=server_meta, + confidentiality=ConfidentialityLabel.PUBLIC, + ) + mcp_tool.session.list_tools.side_effect = [ + SimpleNamespace( + tools=[SimpleNamespace(name="remote_tool", annotations=trusted_annotations)], nextCursor=None + ), + SimpleNamespace( + tools=[SimpleNamespace(name="remote_tool", annotations=untrusted_annotations)], nextCursor=None + ), + ] + proxy = SecureMCPToolProxy(mcp_tool, default_integrity=IntegrityLabel.TRUSTED) + + await proxy.refresh_labels() + props = function.additional_properties + assert props is not None + assert props["source_integrity"] == "trusted" + await proxy.refresh_labels() + assert props["source_integrity"] == "untrusted" + assert function.func is not None + result = await function.func() + + assert result[0].additional_properties["security_label"] == { + "integrity": "untrusted", + "confidentiality": "public", + } + + @pytest.mark.parametrize( + ("start_properties", "refreshed_properties", "first_label", "second_label"), + [ + ( + { + "source_integrity": "trusted", + "confidentiality": "public", + "_mcp_trust_server_ifc": False, + }, + { + "source_integrity": "untrusted", + "confidentiality": "private", + "_mcp_trust_server_ifc": False, + }, + {"integrity": "trusted", "confidentiality": "public"}, + {"integrity": "untrusted", "confidentiality": "private"}, + ), + ( + { + "source_integrity": "untrusted", + "confidentiality": "private", + "_mcp_trust_server_ifc": True, + }, + { + "source_integrity": "untrusted", + "confidentiality": "private", + "_mcp_trust_server_ifc": False, + }, + {"integrity": "trusted", "confidentiality": "public"}, + {"integrity": "untrusted", "confidentiality": "private"}, + ), + ], + ids=["stricter-local-policy", "revoke-server-authority"], + ) + async def test_wrap_mcp_function_snapshots_policy_for_in_flight_call( + self, + start_properties: dict[str, Any], + refreshed_properties: dict[str, Any], + first_label: dict[str, str], + second_label: dict[str, str], + ): + from agent_framework.security import _wrap_mcp_function_for_ifc + + call_started = asyncio.Event() + release_call = asyncio.Event() + + async def fake_call(**kwargs: Any) -> list[Content]: + call_started.set() + await release_call.wait() + return [ + Content.from_text( + "payload", + additional_properties={"_meta": {"ifc": {"integrity": "trusted", "confidentiality": "public"}}}, + ) + ] + + function = FunctionTool( + func=fake_call, + name="remote_tool", + description="", + additional_properties={"_mcp_remote_name": "remote_tool", **start_properties}, + ) + _wrap_mcp_function_for_ifc(function, IntegrityLabel.UNTRUSTED) + assert function.func is not None + + first_call = asyncio.create_task(function.func()) + await call_started.wait() + assert function.additional_properties is not None + function.additional_properties.update(refreshed_properties) + release_call.set() + + first_result = await first_call + second_result = await function.func() + + assert first_result[0].additional_properties["security_label"] == first_label + assert second_result[0].additional_properties["security_label"] == second_label + + @pytest.mark.parametrize("trust_server_ifc", [False, True], ids=["default", "trusted"]) + async def test_apply_mcp_security_labels_configures_result_authority(self, trust_server_ifc: bool): + from agent_framework.security import apply_mcp_security_labels + + annotations = SimpleNamespace(readOnlyHint=True, openWorldHint=False) + server_meta = { + "ifc": {"integrity": "trusted", "confidentiality": "public"}, + "_mcp_trust_server_ifc": True, + } + mcp_tool, function = _make_connected_mcp_tool_for_ifc(annotations=annotations, server_meta=server_meta) + + if trust_server_ifc: + await apply_mcp_security_labels(mcp_tool, trust_server_ifc=True) + else: + await apply_mcp_security_labels(mcp_tool) + + props = function.additional_properties + assert props is not None + assert props["source_integrity"] == "untrusted" + assert props["max_allowed_confidentiality"] == "public" + assert props["accepts_untrusted"] is False + assert props["_mcp_trust_server_ifc"] is trust_server_ifc + assert function.func is not None + result = await function.func() + expected_label = ( + {"integrity": "trusted", "confidentiality": "public"} + if trust_server_ifc + else {"integrity": "untrusted", "confidentiality": "private"} + ) + assert result[0].additional_properties["security_label"] == expected_label + + async def test_apply_mcp_security_labels_reconfigures_existing_wrapper_authority(self): + from agent_framework.security import apply_mcp_security_labels + + annotations = SimpleNamespace(readOnlyHint=True, openWorldHint=False) + server_meta = {"ifc": {"integrity": "trusted", "confidentiality": "public"}} + mcp_tool, function = _make_connected_mcp_tool_for_ifc(annotations=annotations, server_meta=server_meta) + await apply_mcp_security_labels(mcp_tool) + wrapped = function.func + props = function.additional_properties + assert props is not None + assert props["max_allowed_confidentiality"] == "public" + + await apply_mcp_security_labels(mcp_tool, mark_write_tools_as_sinks=False, trust_server_ifc=True) + + assert function.func is wrapped + assert "max_allowed_confidentiality" not in props + assert function.func is not None + result = await function.func() + assert result[0].additional_properties["security_label"] == { + "integrity": "trusted", + "confidentiality": "public", + } + + @pytest.mark.parametrize("trust_server_ifc", [False, True], ids=["default", "trusted"]) + async def test_secure_mcp_proxy_configures_result_authority(self, trust_server_ifc: bool): + from agent_framework.security import SecureMCPToolProxy + + annotations = SimpleNamespace(readOnlyHint=True, openWorldHint=False) + server_meta = {"ifc": {"integrity": "trusted", "confidentiality": "public"}} + mcp_tool, function = _make_connected_mcp_tool_for_ifc(annotations=annotations, server_meta=server_meta) + proxy = ( + SecureMCPToolProxy(mcp_tool, trust_server_ifc=True) if trust_server_ifc else SecureMCPToolProxy(mcp_tool) + ) + + await proxy.refresh_labels() + + props = function.additional_properties + assert props is not None + assert props["source_integrity"] == "untrusted" + assert props["max_allowed_confidentiality"] == "public" + assert props["accepts_untrusted"] is False + assert props["_mcp_trust_server_ifc"] is trust_server_ifc + assert function.func is not None + result = await function.func() + expected_label = ( + {"integrity": "trusted", "confidentiality": "public"} + if trust_server_ifc + else {"integrity": "untrusted", "confidentiality": "private"} + ) + assert result[0].additional_properties["security_label"] == expected_label + async def test_wrap_mcp_function_str_result_passes_through(self): """``str`` results (no per-item containers) are not modified by the wrapper.""" from agent_framework.security import _wrap_mcp_function_for_ifc @@ -5112,7 +5436,6 @@ async def fake_call(**kwargs): result = await func_tool.func() assert result == "plain string result" - @pytest.mark.asyncio async def test_wrap_mcp_function_is_idempotent(self): """Re-running ``_wrap_mcp_function_for_ifc`` (e.g. reconnect) does not double-wrap.""" from agent_framework.security import _wrap_mcp_function_for_ifc diff --git a/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md b/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md index e28d6a6cfc..9ebf7bb19c 100644 --- a/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md +++ b/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md @@ -290,8 +290,8 @@ The middleware now automatically handles variable indirection for UNTRUSTED cont Use it when you need all of the following together: 1. Direct connection to a remote MCP URL from your app process -2. Automatic labeling of tools from MCP annotations (`readOnlyHint`, `openWorldHint`, and related hints) -3. Parsing server result labels from `_meta.ifc` +2. Restriction-only labeling from untrusted MCP annotations (`readOnlyHint`, `openWorldHint`, and related hints) +3. Parsing server result labels from `_meta.ifc` without allowing them to relax local policy by default 4. Local policy enforcement and auto-hide middleware on every tool call **Why this matters:** @@ -363,12 +363,22 @@ async def run_secure_github_mcp(github_pat: str, endpoint: str) -> None: #### What the proxy applies automatically -- Tool metadata labels from MCP hints: +- Restriction-only tool metadata from MCP hints: - `source_integrity` - `accepts_untrusted` - `max_allowed_confidentiality` -- Sink-hardening: non-read-only tools are treated as write-capable and capped to `PUBLIC` confidentiality by default -- Per-result label mapping from `_meta.ifc` into FIDES `security_label` +- Sink-hardening: server annotations cannot remove the `PUBLIC` confidentiality cap or authorize untrusted input +- Per-result label mapping from `_meta.ifc` into FIDES `security_label`; by default, remote labels are combined with + local policy and can only add restrictions + +Set `trust_server_ifc=True` only when the MCP server is an authenticated authority for result labels. In that mode, +a complete valid `_meta.ifc` label is authoritative for that result, including permitted relaxation of the local +fallback. Missing, partial, or malformed labels still use current local policy. This opt-in does not make +ToolAnnotations authoritative: `readOnlyHint` and `openWorldHint` remain restriction-only hints. + +```python +secure_mcp = SecureMCPToolProxy(url="https://trusted.example.com/mcp/", trust_server_ifc=True) +``` #### Operational checklist @@ -377,6 +387,7 @@ async def run_secure_github_mcp(github_pat: str, endpoint: str) -> None: 3. Use `context_providers=[SecureAgentConfig(...)]` instead of manual security wiring. 4. Keep `auto_hide_untrusted=True` unless you have a very specific reason to expose untrusted content. 5. If write-like actions are blocked, inspect `config.get_audit_log(session)` first. +6. Leave `trust_server_ifc=False` unless the connected server is explicitly trusted to label result data. ### 7. Security Tools