From ad70a0a3c232a86748fdb43a1e932d046b8487a5 Mon Sep 17 00:00:00 2001 From: leilei3167 Date: Sat, 26 Sep 2026 22:23:06 +0000 Subject: [PATCH 1/4] Python: block policy violations when approval binding cannot be recorded Fail closed when approval_on_violation is enabled but no approval binding is available, instead of falling through to the permissive allow path. Add regression tests for #8761. --- .../packages/core/agent_framework/security.py | 2 +- python/packages/core/tests/test_security.py | 97 +++++++++++++++++++ 2 files changed, 98 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index b7d87a5f43c..d61cae45580 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -2896,7 +2896,7 @@ async def _process_in_scope( violations=violations, binding=binding, ) - elif self.block_on_violation: + elif self.block_on_violation or self.approval_on_violation: self._block_policy_violation(context, context_label=context_label, violations=violations) else: logger.warning("WARNING: Tool '%s' policy violation(s) [%s] (allowed)", function_name, disclosed) diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 7f371b61a1d..22af8b2b6b1 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -2156,6 +2156,103 @@ async def execute() -> None: assert isinstance(context.result, dict) assert context.result["violation_type"] == "unsafe_approval_binding" + async def test_legacy_user_identity_violation_with_approval_does_not_execute_tool(self) -> None: + """Regression for #8761: approval_on_violation must not allow a violating call.""" + tracker = LabelTrackingFunctionMiddleware() + policy = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) + + async def reads_user_identity() -> str: + return "secret" + + source_tool = FunctionTool( + fn=reads_user_identity, + name="reads_user_identity", + description="Legacy USER_IDENTITY source without principals", + additional_properties={"source_integrity": "trusted", "confidentiality": "user_identity"}, + ) + + async def private_sink(value: str) -> str: + return value + + destination_tool = FunctionTool( + fn=private_sink, + name="microsoft_docs_fetch", + description="Private destination", + additional_properties={"max_allowed_confidentiality": "private"}, + ) + + source_context = FunctionInvocationContext(function=source_tool, arguments={}) + + async def produce_identity(_context: FunctionInvocationContext) -> list[Content]: + return [Content.from_text("secret")] + + await FunctionMiddlewarePipeline(tracker).execute(source_context, produce_identity) + + violation_context = FunctionInvocationContext( + function=destination_tool, + arguments={"value": "x"}, + ) + executed = False + + async def execute(_context: FunctionInvocationContext) -> list[Content]: + nonlocal executed + executed = True + return [Content.from_text("leaked")] + + with pytest.raises(MiddlewareTermination): + await FunctionMiddlewarePipeline(tracker, policy).execute(violation_context, execute) + + assert executed is False + assert isinstance(violation_context.result, dict) + assert violation_context.result["violation_type"] in { + "max_allowed_confidentiality", + "unsafe_approval_binding", + } + + async def test_approval_on_violation_fail_closed_when_binding_record_fails( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + middleware = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) + monkeypatch.setattr(middleware, "_block_unsafe_approval_binding", lambda *args, **kwargs: None) + + class DestinationArgs(BaseModel): + value: str = "value" + + async def destination(value: str = "value") -> str: + return value + + function = FunctionTool( + fn=destination, + name="private_sink", + description="Private destination", + args_schema=DestinationArgs, + additional_properties={"max_allowed_confidentiality": "private"}, + ) + context = FunctionInvocationContext( + function=function, + arguments=function.args_schema(value="value"), # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ) + context.metadata["context_label"] = ContentLabel( + confidentiality=ConfidentialityLabel.USER_IDENTITY, + ) + context.metadata["argument_label"] = ContentLabel() + context.metadata["effective_invocation_label"] = combine_labels( + cast(ContentLabel, context.metadata["context_label"]), + cast(ContentLabel, context.metadata["argument_label"]), + ) + executed = False + + async def execute() -> None: + nonlocal executed + executed = True + + with pytest.raises(MiddlewareTermination): + await middleware.process(context, execute) + + assert executed is False + assert isinstance(context.result, dict) + assert context.result["violation_type"] == "max_allowed_confidentiality" + async def test_approval_binds_computed_argument_principals_without_label_tracker(self) -> None: middleware = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) function = _identity_destination(_principal_metadata("user-c")[_PRINCIPALS_KEY]) From 98777ad51ed33530ab53ac7e8036f7e426c5372c Mon Sep 17 00:00:00 2001 From: leilei3167 Date: Sun, 27 Sep 2026 04:16:38 +0000 Subject: [PATCH 2/4] Python: skip pending approval match when binding record fails Only call _matches_pending_approval after _pending_record succeeds so a mocked or non-terminating unsafe-binding handler cannot hit None binding. --- python/packages/core/agent_framework/security.py | 3 ++- python/packages/core/tests/test_security.py | 5 +++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index d61cae45580..d7c33bb25d1 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -2880,7 +2880,8 @@ async def _process_in_scope( binding = self._pending_record(context, violations) except (TypeError, ValueError, OverflowError): self._block_unsafe_approval_binding(context, context_label=context_label) - approved = self._matches_pending_approval(context, binding) + else: + approved = self._matches_pending_approval(context, binding) disclosed = ", ".join(item["violation_type"] for item in violations) if approved: diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 22af8b2b6b1..9bb23dd2433 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -2215,6 +2215,11 @@ async def test_approval_on_violation_fail_closed_when_binding_record_fails( middleware = PolicyEnforcementFunctionMiddleware(approval_on_violation=True) monkeypatch.setattr(middleware, "_block_unsafe_approval_binding", lambda *args, **kwargs: None) + def fail_pending_record(*_args: object, **_kwargs: object) -> None: + raise ValueError("simulated pending approval binding failure") + + monkeypatch.setattr(middleware, "_pending_record", fail_pending_record) + class DestinationArgs(BaseModel): value: str = "value" From 13c4c44caacd8c54f999a980c120156ec2d47443 Mon Sep 17 00:00:00 2001 From: leilei3167 Date: Mon, 28 Sep 2026 08:11:25 +0000 Subject: [PATCH 3/4] test: remove redundant ContentLabel casts in fail-closed binding test --- python/packages/core/tests/test_security.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 9bb23dd2433..99a7ffaf9b4 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -2242,8 +2242,8 @@ async def destination(value: str = "value") -> str: ) context.metadata["argument_label"] = ContentLabel() context.metadata["effective_invocation_label"] = combine_labels( - cast(ContentLabel, context.metadata["context_label"]), - cast(ContentLabel, context.metadata["argument_label"]), + context.metadata["context_label"], + context.metadata["argument_label"], ) executed = False From 5c54273eed52aee54fa62da59533d3ffe794c2af Mon Sep 17 00:00:00 2001 From: leilei3167 Date: Mon, 28 Sep 2026 09:03:55 +0000 Subject: [PATCH 4/4] Python: raise MiddlewareFailure if unsafe approval binding returns If _block_unsafe_approval_binding returns despite its NoReturn annotation, fail at the call site instead of continuing into the policy ladder. --- python/packages/core/agent_framework/security.py | 5 ++++- python/packages/core/tests/test_security.py | 5 ++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index d7c33bb25d1..85f317b84da 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -34,7 +34,7 @@ from pydantic import BaseModel, Field from ._feature_stage import ExperimentalFeature, experimental -from ._middleware import FunctionInvocationContext, FunctionMiddleware, MiddlewareTermination +from ._middleware import FunctionInvocationContext, FunctionMiddleware, MiddlewareFailure, MiddlewareTermination from ._serialization import SerializationMixin from ._sessions import AgentSession, ContextProvider from ._tools import ( @@ -2880,6 +2880,9 @@ async def _process_in_scope( binding = self._pending_record(context, violations) except (TypeError, ValueError, OverflowError): self._block_unsafe_approval_binding(context, context_label=context_label) + raise MiddlewareFailure( # pyright: ignore[reportUnreachable] + "Unsafe policy approval binding did not terminate" + ) from None else: approved = self._matches_pending_approval(context, binding) diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 99a7ffaf9b4..b2c3aaf82e8 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -2251,12 +2251,11 @@ async def execute() -> None: nonlocal executed executed = True - with pytest.raises(MiddlewareTermination): + with pytest.raises(MiddlewareFailure, match="Unsafe policy approval binding did not terminate"): await middleware.process(context, execute) assert executed is False - assert isinstance(context.result, dict) - assert context.result["violation_type"] == "max_allowed_confidentiality" + assert context.result is None async def test_approval_binds_computed_argument_principals_without_label_tracker(self) -> None: middleware = PolicyEnforcementFunctionMiddleware(approval_on_violation=True)