diff --git a/docs/features/FIDES_IMPLEMENTATION_SUMMARY.md b/docs/features/FIDES_IMPLEMENTATION_SUMMARY.md index db3235bef2..d366c8314e 100644 --- a/docs/features/FIDES_IMPLEMENTATION_SUMMARY.md +++ b/docs/features/FIDES_IMPLEMENTATION_SUMMARY.md @@ -67,7 +67,7 @@ The FIDES defense system consists of seven main components: - Message-level tracking tests (Phase 1) - Data exfiltration prevention tests -4. **`docs/decisions/0011-prompt-injection-defense.md`** +4. **`docs/decisions/0024-prompt-injection-defense.md`** - Architecture Decision Record (ADR) - Design rationale and alternatives considered - Security properties and guarantees diff --git a/python/packages/core/agent_framework/_security.py b/python/packages/core/agent_framework/_security.py index b6e0a535b3..2616b447d3 100644 --- a/python/packages/core/agent_framework/_security.py +++ b/python/packages/core/agent_framework/_security.py @@ -2303,14 +2303,17 @@ inspect_variable(variable_id="var_abc123", reason="Need to determine data format "Use this when you need to process untrusted data (e.g., from external APIs) " "without exposing it to the main conversation. " "You can pass variable_ids directly to reference hidden content from VariableReferenceContent objects. " - "If auto_hide_result is True (default), UNTRUSTED results are automatically hidden." + "UNTRUSTED results are automatically hidden by the middleware." ), additional_properties={ "confidentiality": "private", "accepts_untrusted": True, - # No source_integrity declared: middleware falls back to Tier 3 - # (join of input argument labels), so output inherits trust from - # inputs — matching the tool's internal combine_labels() logic. + "source_integrity": "untrusted", + # source_integrity is declared as UNTRUSTED because this tool + # processes external/untrusted data. The middleware uses this + # (Tier 2) to label the output UNTRUSTED and auto-hide it via + # the standard _should_hide() → _hide_item() path — no + # tool-internal auto-hide logic needed. }, ) async def quarantined_llm( @@ -2324,27 +2327,23 @@ async def quarantined_llm( Field(description="Dictionary of labeled data items (alternative to variable_ids)"), ] = None, metadata: Annotated[dict[str, Any] | None, Field(description="Optional metadata")] = None, - auto_hide_result: Annotated[ - bool, - Field(description="If True, automatically hide UNTRUSTED results in variable store"), - ] = True, ) -> dict[str, Any]: """Make an isolated LLM call with labeled data. This tool creates a quarantined LLM context where untrusted content can be processed - without exposing it to the main agent conversation. The result is labeled with - the combined security labels of all inputs. + without exposing it to the main agent conversation. The result is labeled as UNTRUSTED + via the tool's ``source_integrity`` declaration, and the middleware automatically hides + it behind a variable reference when ``auto_hide_untrusted`` is enabled. Args: prompt: The prompt to send to the quarantined LLM. variable_ids: List of variable IDs to retrieve and process from the variable store. labelled_data: Dictionary of labeled data items with their security labels. metadata: Optional additional metadata for the request. - auto_hide_result: Whether to automatically hide UNTRUSTED results in the variable store. Returns: Dictionary containing: - - response: The LLM's response (placeholder in this implementation) + - response: The LLM's response - security_label: The combined security label - metadata: Request metadata - variables_processed: List of variable IDs that were processed @@ -2511,46 +2510,21 @@ async def quarantined_llm( logger.warning("No quarantine client configured, using placeholder response") response_text = f"[Quarantined LLM Response] Processed: {prompt[:100]}" - # Handle auto_hide_result parameter - actual_auto_hide = auto_hide_result - - # If result is UNTRUSTED and auto_hide is enabled, store in variable and return reference - if actual_auto_hide and combined_label.integrity == IntegrityLabel.UNTRUSTED: - # Store the actual response in variable store - var_id = variable_store.store(response_text, combined_label) - - logger.info( - f"Quarantined LLM result auto-hidden in variable {var_id} (label: {combined_label.integrity.value})" - ) - - # Return a VariableReferenceContent-style response - response_payload: dict[str, Any] = { - "type": "variable_reference", - "variable_id": var_id, - "description": f"Quarantined LLM result (derived from {len(actual_variable_ids)} sources)", - "security_label": combined_label.to_dict(), - "metadata": actual_metadata or {}, - "quarantined": True, - "auto_hidden": True, - "variables_processed": list(actual_variable_ids), - "content_summary": content_summary, - } - else: - # Return the response directly (TRUSTED or auto_hide disabled) - response_payload = { - "response": response_text, - "security_label": combined_label.to_dict(), - "metadata": actual_metadata or {}, - "quarantined": True, - "auto_hidden": False, - "variables_processed": list(actual_variable_ids), - "content_summary": content_summary, - } + # Return the response — the middleware's _label_result() will handle + # auto-hiding via _should_hide() → _hide_item() based on the tool's + # source_integrity="untrusted" declaration. + response_payload: dict[str, Any] = { + "response": response_text, + "security_label": combined_label.to_dict(), + "metadata": actual_metadata or {}, + "quarantined": True, + "variables_processed": list(actual_variable_ids), + "content_summary": content_summary, + } logger.info( f"Quarantined LLM response generated with label: " - f"{combined_label.integrity.value}, {combined_label.confidentiality.value}, " - f"auto_hidden={response_payload.get('auto_hidden', False)}" + f"{combined_label.integrity.value}, {combined_label.confidentiality.value}" ) return response_payload @@ -2576,12 +2550,14 @@ class InspectVariableInput(BaseModel): "prompt injection attempts. Only use when absolutely necessary and with caution. " "The context label will be marked as UNTRUSTED after inspection." ), - approval_mode="always_require", + approval_mode="never_require", additional_properties={ "confidentiality": "private", # No source_integrity declared: output inherits the label of the # inspected content via Tier 3. The variable store is just a # container — the data inside it is untrusted external content. + # No approval_mode gate: inspect_variable runs freely but taints the + # context to UNTRUSTED, which blocks dangerous tools via policy. }, ) async def inspect_variable( diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 1fe11e2090..1d4dba554a 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -1590,12 +1590,20 @@ async def _auto_invoke_function( except MiddlewareTermination as term_exc: # Re-raise to signal loop termination, but first capture any result set by middleware if middleware_context.result is not None: - # Store result in exception for caller to extract - term_exc.result = Content.from_function_result( - call_id=call_id, - result=middleware_context.result, - additional_properties=function_call_content.additional_properties, - ) + # Pass through function_approval_request directly (e.g., from security policy middleware) + # so the approval flow in _handle_function_call_results activates correctly. + if ( + isinstance(middleware_context.result, Content) + and middleware_context.result.type == "function_approval_request" + ): + term_exc.result = middleware_context.result + else: + # Store result in exception for caller to extract + term_exc.result = Content.from_function_result( + call_id=call_id, + result=middleware_context.result, + additional_properties=function_call_content.additional_properties, + ) raise except UserInputRequiredException: raise diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index a511fe9e1e..c735244603 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -645,6 +645,47 @@ class TestPolicyEnforcementMiddleware: assert captured_metadata["approval_response"] is approval_response assert captured_metadata["policy_approval_granted"] is None + async def test_policy_violation_approval_preserves_type_through_auto_invoke(self, mock_function): + """Test that _auto_invoke_function preserves function_approval_request type on MiddlewareTermination. + + When PolicyEnforcementFunctionMiddleware raises MiddlewareTermination with a + function_approval_request result, the exception handler must pass it through + directly rather than wrapping it in a function_result. + """ + label_tracker = LabelTrackingFunctionMiddleware(auto_hide_untrusted=False) + # Taint the context label so the policy enforcer sees UNTRUSTED + label_tracker._context_label = ContentLabel(integrity=IntegrityLabel.UNTRUSTED) + label_tracker._initialized = True + + policy = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) + pipeline = FunctionMiddlewarePipeline(label_tracker, policy) + + function_call = Content.from_function_call( + call_id="call-policy-violation", + name=mock_function.name, + arguments='{"arg": "test"}', + ) + + with pytest.raises(MiddlewareTermination) as exc_info: + await _auto_invoke_function( + function_call, + config=normalize_function_invocation_configuration(None), + tool_map={mock_function.name: mock_function}, + middleware_pipeline=pipeline, + ) + + # The exception's result must be a function_approval_request, NOT a function_result + result = exc_info.value.result + assert isinstance(result, Content) + assert result.type == "function_approval_request", ( + f"Expected function_approval_request but got {result.type}; " + "MiddlewareTermination handler must not wrap approval requests in function_result" + ) + assert result.function_call is not None + assert result.function_call.call_id == "call-policy-violation" + assert result.additional_properties["policy_violation"] is True + assert result.additional_properties["violation_type"] == "untrusted_context" + class TestAutomaticHiding: """Tests for automatic variable hiding functionality.""" @@ -908,11 +949,11 @@ class TestSecureAgentConfig: assert "inspect_variable" in instructions def test_inspect_variable_uses_generic_approval_mode(self): - """Test that inspect_variable uses the standard tool approval flow.""" + """Test that inspect_variable does not require approval (context tainting handles security).""" from agent_framework import get_security_tools inspect_variable = next(tool for tool in get_security_tools() if tool.name == "inspect_variable") - assert inspect_variable.approval_mode == "always_require" + assert inspect_variable.approval_mode == "never_require" assert "requires_approval" not in inspect_variable.additional_properties @@ -1480,15 +1521,19 @@ class TestMiddlewareMessageLabeling: assert len(middleware.get_all_message_labels()) == 0 -# ========== Quarantined LLM Auto-Hide Tests ========== +# ========== Quarantined LLM Tests ========== -class TestQuarantinedLLMAutoHide: - """Tests for quarantined_llm auto-hiding of UNTRUSTED results.""" +class TestQuarantinedLLM: + """Tests for quarantined_llm tool behavior. + + Note: Auto-hiding of UNTRUSTED results is handled by the middleware + via source_integrity="untrusted", not by quarantined_llm itself. + """ @pytest.mark.asyncio - async def test_quarantined_llm_auto_hides_untrusted_result(self): - """Test that quarantined_llm auto-hides UNTRUSTED results.""" + async def test_quarantined_llm_returns_response(self): + """Test that quarantined_llm returns a plain response dict.""" from agent_framework import LabelTrackingFunctionMiddleware, quarantined_llm from agent_framework._security import _current_middleware @@ -1503,49 +1548,24 @@ class TestQuarantinedLLMAutoHide: _current_middleware.instance = middleware try: - result = await quarantined_llm(prompt="Summarize this data", variable_ids=[var_id], auto_hide_result=True) + result = await quarantined_llm(prompt="Summarize this data", variable_ids=[var_id]) - # Result should be auto-hidden since input was UNTRUSTED - assert result["auto_hidden"] is True - assert result["type"] == "variable_reference" - assert "variable_id" in result - assert result["variable_id"].startswith("var_") - finally: - _current_middleware.instance = None - - @pytest.mark.asyncio - async def test_quarantined_llm_no_hide_when_disabled(self): - """Test that auto_hide_result=False prevents hiding.""" - from agent_framework import LabelTrackingFunctionMiddleware, quarantined_llm - from agent_framework._security import _current_middleware - - middleware = LabelTrackingFunctionMiddleware() - - var_id = middleware.get_variable_store().store( - "untrusted data", ContentLabel(integrity=IntegrityLabel.UNTRUSTED) - ) - - _current_middleware.instance = middleware - - try: - result = await quarantined_llm(prompt="Process this", variable_ids=[var_id], auto_hide_result=False) - - # Result should NOT be hidden - assert result["auto_hidden"] is False + # Result should be a plain response dict (middleware handles hiding) assert "response" in result - assert "type" not in result or result.get("type") != "variable_reference" + assert result["quarantined"] is True + assert "auto_hidden" not in result finally: _current_middleware.instance = None @pytest.mark.asyncio - async def test_quarantined_llm_trusted_result_not_hidden(self): - """Test that TRUSTED results are not auto-hidden.""" + async def test_quarantined_llm_trusted_input(self): + """Test quarantined_llm with TRUSTED input returns response directly.""" from agent_framework import LabelTrackingFunctionMiddleware, quarantined_llm from agent_framework._security import _current_middleware middleware = LabelTrackingFunctionMiddleware() - # Store TRUSTED content (unusual but possible) + # Store TRUSTED content var_id = middleware.get_variable_store().store( "trusted system data", ContentLabel(integrity=IntegrityLabel.TRUSTED) ) @@ -1556,12 +1576,11 @@ class TestQuarantinedLLMAutoHide: result = await quarantined_llm( prompt="Process this", variable_ids=[var_id], - auto_hide_result=True, # Still enabled ) - # Result should NOT be hidden because input was TRUSTED - assert result["auto_hidden"] is False + # Result should be a plain response dict assert "response" in result + assert result["quarantined"] is True finally: _current_middleware.instance = None @@ -1587,6 +1606,14 @@ class TestQuarantinedLLMAutoHide: finally: _current_middleware.instance = None + def test_quarantined_llm_declares_source_integrity(self): + """Test that quarantined_llm declares source_integrity='untrusted'.""" + from agent_framework import get_security_tools + + q_llm = next(tool for tool in get_security_tools() if tool.name == "quarantined_llm") + assert q_llm.additional_properties.get("source_integrity") == "untrusted" + assert q_llm.additional_properties.get("accepts_untrusted") is True + class TestQuarantineClient: """Tests for quarantine chat client functionality.""" @@ -1709,9 +1736,9 @@ class TestQuarantineClient: assert call_args.kwargs.get("tools") is None assert call_args.kwargs.get("client_kwargs", {}).get("tool_choice") == "none" - # Since it's untrusted and auto_hide is True, result should be hidden - assert result["auto_hidden"] is True - assert "variable_id" in result + # Result should be a plain response dict (middleware handles hiding) + assert "response" in result + assert result["response"] == "This is a safe summary of the content." finally: _current_middleware.instance = None @@ -1744,7 +1771,6 @@ class TestQuarantineClient: result = await quarantined_llm( prompt="Process this content", variable_ids=[var_id], - auto_hide_result=False, # Disable auto-hide to see the response ) # Should use placeholder response @@ -1780,7 +1806,7 @@ class TestQuarantineClient: _current_middleware.instance = middleware try: - result = await quarantined_llm(prompt="Process this", variable_ids=[var_id], auto_hide_result=False) + result = await quarantined_llm(prompt="Process this", variable_ids=[var_id]) # Should fall back to error message assert "response" in result diff --git a/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md b/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md index 15eea48ad2..6e7abbba63 100644 --- a/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md +++ b/python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md @@ -372,22 +372,13 @@ result = await quarantined_llm( } } ) - -# Option 3: Auto-hide results (default behavior for UNTRUSTED inputs) -result = await quarantined_llm( - prompt="Process this", - variable_ids=["var_abc123"], - auto_hide_result=True # Default: hides result if inputs are UNTRUSTED -) -# Returns variable reference instead of raw response ``` **Key Security Features:** - Content is processed with `tools=None` and `tool_choice="none"` - Prompt injection attempts in the content cannot trigger tool calls -- Results inherit the most restrictive label from inputs -- UNTRUSTED results are automatically hidden (stored as variable references) -``` +- Declares `source_integrity="untrusted"` — the middleware automatically hides results via the standard auto-hide mechanism +- No tool-internal auto-hide logic — hiding is handled uniformly by `LabelTrackingFunctionMiddleware` #### inspect_variable @@ -407,9 +398,12 @@ async def inspect_content() -> None: # WARNING: Exposes untrusted content to context ``` -`inspect_variable` uses the standard tool approval flow via `approval_mode="always_require"`. -That is separate from secure-policy approvals triggered by `SecureAgentConfig(..., approval_on_violation=True)`, -which only request approval when a call would otherwise be blocked by the current security context. +`inspect_variable` uses `approval_mode="never_require"` because the tool call is internal to the +security framework and not visible to the developer. Instead of gating on approval, calling +`inspect_variable` taints the context to UNTRUSTED, which blocks dangerous tool calls via +`PolicyEnforcementFunctionMiddleware`. This is separate from secure-policy approvals triggered +by `SecureAgentConfig(..., approval_on_violation=True)`, which only request approval when a +call would otherwise be blocked by the current security context. ### 7. SecureAgentConfig (Context Provider) @@ -551,30 +545,20 @@ msg = LabeledMessage( **quarantined_llm Auto-Hiding:** -`quarantined_llm` automatically hides UNTRUSTED results: +`quarantined_llm` declares `source_integrity="untrusted"` in its tool metadata. The +`LabelTrackingFunctionMiddleware` uses this to label the output as UNTRUSTED and +automatically hide it behind a variable reference — the same mechanism used for any +other tool that returns untrusted data. No tool-internal auto-hide logic is needed. ```python -# When processing UNTRUSTED content, result is auto-hidden +# When processing UNTRUSTED content, the middleware auto-hides the result result = await quarantined_llm( prompt="Summarize this data", - variable_ids=["var_abc123"], - auto_hide_result=True # Default: True -) - -# If input was UNTRUSTED, result is: -# { -# "type": "variable_reference", -# "variable_id": "var_xyz789", # Auto-hidden result -# "auto_hidden": True, -# ... -# } - -# Disable auto-hiding if needed -result = await quarantined_llm( - prompt="Process this", - variable_ids=["var_abc123"], - auto_hide_result=False # Return response directly + variable_ids=["var_abc123"] ) +# The middleware stores the response in the variable store and replaces it +# with a VariableReferenceContent — just like any other untrusted tool result. +# The agent can then use inspect_variable() to surface the content. ``` ## Usage Examples @@ -1158,30 +1142,20 @@ result = await quarantined_llm( variable_ids: List[str] = [], # Variable IDs to retrieve from store labelled_data: Dict[str, Any] = {}, # Alternative: direct labeled data metadata: Dict[str, Any] = None, # Optional metadata - auto_hide_result: bool = True, # Auto-hide UNTRUSTED results (NEW!) ) -> Dict[str, Any] -# Returns (when auto_hidden=False or result is TRUSTED): +# Returns: # { # "response": str, # LLM response # "security_label": dict, # Combined label of all inputs # "quarantined": True, -# "auto_hidden": False, -# "variables_processed": List[str], -# "content_summary": List[str], -# } - -# Returns (when auto_hidden=True AND result is UNTRUSTED): -# { -# "type": "variable_reference", -# "variable_id": str, # ID of auto-hidden result -# "description": str, -# "security_label": dict, -# "quarantined": True, -# "auto_hidden": True, # "variables_processed": List[str], # "content_summary": List[str], # } +# +# Note: The middleware automatically hides UNTRUSTED results behind a +# VariableReferenceContent via the tool's source_integrity="untrusted" +# declaration. The agent sees a variable reference, not raw content. ``` ### inspect_variable diff --git a/python/samples/02-agents/security/README.md b/python/samples/02-agents/security/README.md index 056a7edaa9..c43b1ecb4d 100644 --- a/python/samples/02-agents/security/README.md +++ b/python/samples/02-agents/security/README.md @@ -461,10 +461,10 @@ Run the security examples: cd python # Email security (prompt injection defense) -PYTHONPATH=packages/core python samples/getting_started/security/email_security_example.py +PYTHONPATH=packages/core python samples/02-agents/security/email_security_example.py # Repository confidentiality (data exfiltration prevention) -PYTHONPATH=packages/core python samples/getting_started/security/repo_confidentiality_example.py +PYTHONPATH=packages/core python samples/02-agents/security/repo_confidentiality_example.py ``` These show: @@ -477,10 +477,10 @@ These show: ## More Information -- Full documentation: `python/packages/core/FIDES_DEVELOPER_GUIDE.md` +- Full documentation: `python/samples/02-agents/security/FIDES_DEVELOPER_GUIDE.md` - Test suite: `python/packages/core/tests/test_security.py` -- Email example: `python/samples/getting_started/security/email_security_example.py` -- Repo example: `python/samples/getting_started/security/repo_confidentiality_example.py` +- Email example: `python/samples/02-agents/security/email_security_example.py` +- Repo example: `python/samples/02-agents/security/repo_confidentiality_example.py` ## Support