diff --git a/python/samples/concepts/README.md b/python/samples/concepts/README.md index bfde792aed39..072dd8a9da34 100644 --- a/python/samples/concepts/README.md +++ b/python/samples/concepts/README.md @@ -150,6 +150,7 @@ ### Filtering - Creating and using Filters - [Auto Function Invoke Filters](./filtering/auto_function_invoke_filters.py) +- [Function Authorization Filter](./filtering/function_authorization_filter.py) - [Function Invocation Filters](./filtering/function_invocation_filters.py) - [Function Invocation Filters Stream](./filtering/function_invocation_filters_stream.py) - [Prompt Filters](./filtering/prompt_filters.py) diff --git a/python/samples/concepts/filtering/function_authorization_filter.py b/python/samples/concepts/filtering/function_authorization_filter.py new file mode 100644 index 000000000000..fb412b62e46a --- /dev/null +++ b/python/samples/concepts/filtering/function_authorization_filter.py @@ -0,0 +1,121 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio + +from semantic_kernel import Kernel +from semantic_kernel.contents import ChatHistory +from semantic_kernel.contents.function_call_content import FunctionCallContent +from semantic_kernel.filters import ( + FilterTypes, + FunctionAuthorizationFilter, + FunctionAuthorizationPolicy, + FunctionRiskLevel, +) +from semantic_kernel.functions import kernel_function + +""" +This sample shows how to gate auto function invocation behind an explicit, +auditable authorization decision using the FunctionAuthorizationFilter +(see https://github.com/microsoft/semantic-kernel/issues/14072). + +The scenario: an agent has both a harmless read-only function and a +destructive one. An indirect prompt injection (e.g. hidden instructions in a +retrieved document) can trick the model into *proposing* a destructive tool +call — but with the authorization filter registered, proposing a call is no +longer the same as executing it: + +- low-risk calls are dispatched as usual; +- the destructive call is suspended as a pending decision and never runs; +- a human grants the pending decision, and re-issuing the *identical* call + executes exactly once; +- replaying that approval with tampered arguments is rejected, because the + approval is bound to the canonical argument digest. + +The model side is simulated with hand-built FunctionCallContent objects, so +the sample runs without any model API key: kernel.invoke_function_call() is +exactly the entry point a chat completion service uses for every tool call +the model proposes during auto function invocation. +""" + + +class FileSystemPlugin: + """A plugin exposing a harmless function and a destructive one.""" + + def __init__(self): + self.deleted: list[str] = [] + + @kernel_function(name="read_file", description="Read a file from the workspace.") + def read_file(self, path: str) -> str: + return f"contents of {path}" + + @kernel_function(name="delete_path", description="Delete a file or directory tree.") + def delete_path(self, path: str) -> str: + self.deleted.append(path) + return f"deleted {path}" + + +async def main() -> None: + kernel = Kernel() + file_system = FileSystemPlugin() + kernel.add_plugin(file_system, plugin_name="fs") + + # Declare risk in function metadata (the filter also supports policy-side + # overrides, and fails closed to HIGH for anything left unclassified). + kernel.get_function("fs", "read_file").metadata.additional_properties = {"risk_level": "low"} + kernel.get_function("fs", "delete_path").metadata.additional_properties = {"risk_level": "high"} + + auth_filter = FunctionAuthorizationFilter( + policy=FunctionAuthorizationPolicy( + principal="demo_user", + # A deterministic tripwire: suspicious argument content escalates + # the risk before dispatch, whatever the function's declared risk. + keyword_guard={"..": FunctionRiskLevel.CRITICAL}, + ) + ) + kernel.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, auth_filter) + + history = ChatHistory() + + async def model_proposes(call_id: str, function_name: str, arguments: str): + """Stand-in for the model's tool call during auto function invocation.""" + print(f"\nModel proposes: {function_name}({arguments})") + await kernel.invoke_function_call( + function_call=FunctionCallContent( + id=call_id, plugin_name="fs", function_name=function_name, arguments=arguments + ), + chat_history=history, + ) + print(f" -> fed back to the model: {history.messages[-1].items[0].result}") + + # 1. A benign, low-risk call is dispatched as usual. + await model_proposes("call_1", "read_file", '{"path": "report.md"}') + + # 2. An indirect prompt injection tricks the model into proposing a + # destructive call. The filter suspends it: nothing is deleted. + await model_proposes("call_2", "delete_path", '{"path": "workspace/archive"}') + pending = auth_filter.audit_log[-1] + print(f" deleted so far: {file_system.deleted} (decision: {pending.status.value})") + + # 3. A human reviews the pending decision and grants it, then the caller + # re-issues the identical call: it now executes exactly once. + auth_filter.grant_approval(pending) + await model_proposes("call_3", "delete_path", '{"path": "workspace/archive"}') + print(f" deleted so far: {file_system.deleted}") + + # 4. Replaying with tampered arguments fails twice over: the earlier + # approval was bound to the exact argument digest (and was consumed), + # and the path-traversal payload trips the keyword guard, which + # escalates the call to CRITICAL and denies it outright. + await model_proposes("call_4", "delete_path", '{"path": "workspace/../production"}') + print(f" deleted so far: {file_system.deleted}") + + print("\nAudit trail:") + for decision in auth_filter.audit_log: + print( + f" [{decision.status.value:>16}] {decision.function_name} " + f"risk={decision.risk.value} via {decision.authority_source}: {decision.reason}" + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/semantic_kernel/filters/__init__.py b/python/semantic_kernel/filters/__init__.py index b0b8b16e35c9..17f8f4d12b7d 100644 --- a/python/semantic_kernel/filters/__init__.py +++ b/python/semantic_kernel/filters/__init__.py @@ -3,6 +3,15 @@ from semantic_kernel.filters.auto_function_invocation.auto_function_invocation_context import ( AutoFunctionInvocationContext, ) +from semantic_kernel.filters.auto_function_invocation.function_authorization_filter import ( + FunctionApprovalStore, + FunctionAuthorizationAction, + FunctionAuthorizationDecision, + FunctionAuthorizationFilter, + FunctionAuthorizationPolicy, + FunctionAuthorizationStatus, + FunctionRiskLevel, +) from semantic_kernel.filters.filter_types import FilterTypes from semantic_kernel.filters.functions.function_invocation_context import FunctionInvocationContext from semantic_kernel.filters.prompts.prompt_render_context import PromptRenderContext @@ -10,6 +19,13 @@ __all__ = [ "AutoFunctionInvocationContext", "FilterTypes", + "FunctionApprovalStore", + "FunctionAuthorizationAction", + "FunctionAuthorizationDecision", + "FunctionAuthorizationFilter", + "FunctionAuthorizationPolicy", + "FunctionAuthorizationStatus", "FunctionInvocationContext", + "FunctionRiskLevel", "PromptRenderContext", ] diff --git a/python/semantic_kernel/filters/auto_function_invocation/function_authorization_filter.py b/python/semantic_kernel/filters/auto_function_invocation/function_authorization_filter.py new file mode 100644 index 000000000000..4dc84a7cc1ee --- /dev/null +++ b/python/semantic_kernel/filters/auto_function_invocation/function_authorization_filter.py @@ -0,0 +1,472 @@ +# Copyright (c) Microsoft. All rights reserved. + +import hashlib +import json +import logging +import math +import time +import uuid +from collections.abc import Mapping +from enum import Enum +from typing import TYPE_CHECKING, Any + +from pydantic import Field + +from semantic_kernel.functions.function_result import FunctionResult +from semantic_kernel.kernel_pydantic import KernelBaseModel +from semantic_kernel.utils.feature_stage_decorator import experimental + +if TYPE_CHECKING: + from semantic_kernel.filters.auto_function_invocation.auto_function_invocation_context import ( + AutoFunctionInvocationContext, + ) + +logger = logging.getLogger(__name__) + +RISK_LEVEL_PROPERTY = "risk_level" +REQUIRES_APPROVAL_PROPERTY = "requires_approval" + + +@experimental +class FunctionRiskLevel(str, Enum): + """Risk classification for a kernel function used by the authorization filter.""" + + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + CRITICAL = "critical" + + +_RISK_ORDER = { + FunctionRiskLevel.LOW: 0, + FunctionRiskLevel.MEDIUM: 1, + FunctionRiskLevel.HIGH: 2, + FunctionRiskLevel.CRITICAL: 3, +} + + +@experimental +class FunctionAuthorizationAction(str, Enum): + """Terminal authorization action for a proposed auto function invocation.""" + + ALLOW = "allow" + DENY = "deny" + REQUIRE_APPROVAL = "require_approval" + + +@experimental +class FunctionAuthorizationStatus(str, Enum): + """Lifecycle status of an authorization decision, recorded in the audit log.""" + + PENDING_APPROVAL = "pending_approval" + APPROVED = "approved" + DENIED = "denied" + EXPIRED = "expired" + EXECUTED = "executed" + FAILED = "failed" + + +@experimental +class FunctionAuthorizationDecision(KernelBaseModel): + """An explicit, auditable record of a single authorization decision. + + The decision binds the outcome to the exact call that was proposed: + the fully qualified function name, a canonical digest of the arguments, + the principal on whose behalf the call runs and a digest of the policy + that produced the decision. An approval is only valid for the identical + binding, so changing any of these invalidates it. + """ + + decision_id: str = Field(default_factory=lambda: uuid.uuid4().hex) + function_name: str + args_digest: str + principal: str + policy_digest: str + risk: FunctionRiskLevel + action: FunctionAuthorizationAction + status: FunctionAuthorizationStatus + reason: str + authority_source: str + created_at: float = Field(default_factory=time.time) + + @property + def binding(self) -> str: + """The tag an approval must match for this exact call to be dispatched.""" + material = "|".join([self.function_name, self.args_digest, self.principal, self.policy_digest]) + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + +@experimental +class FunctionApprovalStore: + """In-memory, single-use store of granted approvals keyed by decision binding. + + Approvals expire after a time-to-live and are consumed on use, so a granted + approval authorizes at most one dispatch of the exact call it was bound to. + Replace with a durable implementation for multi-process scenarios. + """ + + def __init__(self) -> None: + """Initialize an empty approval store.""" + self._approvals: dict[str, float] = {} + + def grant(self, binding: str, ttl_seconds: float) -> None: + """Grant a single-use approval for the given decision binding. + + Raises: + ValueError: If ``ttl_seconds`` is not a finite, positive number, + which would otherwise create an approval that never expires. + """ + if not math.isfinite(ttl_seconds) or ttl_seconds <= 0: + raise ValueError(f"ttl_seconds must be a finite, positive number, got {ttl_seconds!r}.") + self._approvals[binding] = time.time() + ttl_seconds + + def consume(self, binding: str) -> str: + """Consume an approval for the binding; returns 'ok', 'expired' or 'absent'.""" + expires_at = self._approvals.pop(binding, None) + if expires_at is None: + return "absent" + if time.time() > expires_at: + return "expired" + return "ok" + + +@experimental +class FunctionAuthorizationPolicy(KernelBaseModel): + """Declarative, deterministic policy evaluated before every auto function invocation. + + Risk resolution is fail-closed: a function with no declared risk is treated as + ``default_risk`` (HIGH unless configured otherwise), and the most restrictive + signal wins between declared risk and the keyword guard. + + Args: + risk_overrides: Map of fully qualified function name (``plugin-function``) + or plugin wildcard (``plugin-*``) to a risk level. Exact names win + over wildcards. + keyword_guard: Case-insensitive substrings that escalate the resolved + risk to at least the mapped level when found in the rendered + arguments. This is a deterministic tripwire for indirect prompt + injection payloads, not a semantic classifier. + action_map: Terminal action per risk level. Defaults to ALLOW for + LOW/MEDIUM, REQUIRE_APPROVAL for HIGH and DENY for CRITICAL. + default_risk: Risk assigned to functions with no declared classification. + principal: Identifier of the caller/session the decisions are bound to. + approval_ttl_seconds: Lifetime of a granted approval. + terminate_on_pending: Whether a pending approval also terminates the + auto-invocation loop so the pending decision surfaces to the caller. + """ + + risk_overrides: dict[str, FunctionRiskLevel] = Field(default_factory=dict) + keyword_guard: dict[str, FunctionRiskLevel] = Field(default_factory=dict) + action_map: dict[FunctionRiskLevel, FunctionAuthorizationAction] = Field( + default_factory=lambda: { + FunctionRiskLevel.LOW: FunctionAuthorizationAction.ALLOW, + FunctionRiskLevel.MEDIUM: FunctionAuthorizationAction.ALLOW, + FunctionRiskLevel.HIGH: FunctionAuthorizationAction.REQUIRE_APPROVAL, + FunctionRiskLevel.CRITICAL: FunctionAuthorizationAction.DENY, + } + ) + default_risk: FunctionRiskLevel = FunctionRiskLevel.HIGH + principal: str = "default" + approval_ttl_seconds: float = 300.0 + terminate_on_pending: bool = True + + @property + def policy_digest(self) -> str: + """A digest of the policy, so approvals are invalidated by policy changes.""" + snapshot = json.dumps( + { + "risk_overrides": {k: v.value for k, v in sorted(self.risk_overrides.items())}, + "keyword_guard": {k: v.value for k, v in sorted(self.keyword_guard.items())}, + "action_map": {k.value: v.value for k, v in sorted(self.action_map.items())}, + "default_risk": self.default_risk.value, + "principal": self.principal, + }, + sort_keys=True, + ) + return hashlib.sha256(snapshot.encode("utf-8")).hexdigest() + + def resolve_risk( + self, + fully_qualified_name: str, + plugin_name: str | None, + additional_properties: Mapping[str, Any] | None, + rendered_arguments: str, + ) -> tuple[FunctionRiskLevel, str, bool]: + """Resolve the risk for a proposed call; returns (risk, reason, requires_approval). + + Precedence: exact policy override > plugin wildcard override > function + metadata (``additional_properties``) > ``default_risk`` (fail-closed). + The keyword guard can only escalate the resolved risk, never lower it. + """ + requires_approval = False + if fully_qualified_name in self.risk_overrides: + risk = self.risk_overrides[fully_qualified_name] + reason = f"policy override for '{fully_qualified_name}'" + elif plugin_name and f"{plugin_name}-*" in self.risk_overrides: + risk = self.risk_overrides[f"{plugin_name}-*"] + reason = f"policy override for plugin '{plugin_name}'" + else: + declared = (additional_properties or {}).get(RISK_LEVEL_PROPERTY) + if declared is not None: + try: + risk = FunctionRiskLevel(str(declared).lower()) + reason = f"declared {RISK_LEVEL_PROPERTY}='{risk.value}' in function metadata" + except ValueError: + risk = self.default_risk + reason = ( + f"invalid {RISK_LEVEL_PROPERTY} '{declared}' in function metadata, " + f"failing closed to default risk '{risk.value}'" + ) + else: + risk = self.default_risk + reason = f"no declared risk, failing closed to default risk '{risk.value}'" + if (additional_properties or {}).get(REQUIRES_APPROVAL_PROPERTY): + requires_approval = True + haystack = rendered_arguments.lower() + for keyword, guard_risk in self.keyword_guard.items(): + if keyword.lower() in haystack and _RISK_ORDER[guard_risk] > _RISK_ORDER[risk]: + risk = guard_risk + reason = f"keyword guard matched '{keyword}', escalating to '{risk.value}'" + return risk, reason, requires_approval + + def action_for(self, risk: FunctionRiskLevel, requires_approval: bool) -> FunctionAuthorizationAction: + """Map resolved risk to a terminal action; unmapped risk levels fail closed to DENY.""" + action = self.action_map.get(risk, FunctionAuthorizationAction.DENY) + if requires_approval and action == FunctionAuthorizationAction.ALLOW: + return FunctionAuthorizationAction.REQUIRE_APPROVAL + return action + + +@experimental +class FunctionAuthorizationFilter: + """An AUTO_FUNCTION_INVOCATION filter that turns function dispatch into an authorized action. + + Register it like any other filter:: + + kernel.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, auth_filter) + + For every function call proposed by the model, the filter deterministically + resolves a risk level and records an explicit :class:`FunctionAuthorizationDecision`: + + - ALLOW: the call is dispatched (``await next(context)``). + - DENY: the call is never dispatched; the model receives a structured refusal. + - REQUIRE_APPROVAL: the call is never dispatched; a pending decision is + surfaced as the function result (and, by default, terminates the + auto-invocation loop). After :meth:`grant_approval`, re-invoking the same + call — same function, same canonical arguments, same principal, same + policy — dispatches it exactly once. + + Because approvals are bound to the canonical argument digest, an approval + granted for ``transfer(amount=10)`` can never be replayed for + ``transfer(amount=10000)``, closing the time-of-check/time-of-use gap. + Prompt injection can therefore make the model *propose* a call, but never + bypass the checkpoint that dispatches it. + + Notes: + - Filters form a chain and the most recently added filter runs + innermost. Add this filter *last*, so no other filter runs (and can + mutate the arguments) between the authorization check and dispatch. + - Filters are trusted application code. The boundary enforced here is + between untrusted model output and dispatch, not between filters. + - EXECUTED means the call was dispatched; if dispatch raises, the + decision is recorded as FAILED and the exception propagates. + - Arguments that cannot be canonicalized fail closed to DENY. + """ + + def __init__( + self, + policy: FunctionAuthorizationPolicy | None = None, + approval_store: FunctionApprovalStore | None = None, + ) -> None: + """Initialize the filter with an optional policy and approval store.""" + self.policy = policy or FunctionAuthorizationPolicy() + self.approvals = approval_store or FunctionApprovalStore() + self.audit_log: list[FunctionAuthorizationDecision] = [] + self._ordering_warned = False + + @classmethod + def render_canonical_arguments(cls, arguments: Mapping[str, Any] | None) -> str: + """Render the call arguments to a canonical, order-independent string. + + Every value is encoded together with a structural type tag, so values + of different types can never render identically: a native string can + never collide with an object whose ``__str__`` (or type-prefixed + rendering) mimics it, because their tags differ. + + Any failure to canonicalize (for example, a circular reference) raises, + and callers must treat it as fail-closed. + """ + return json.dumps(cls._canonicalize(dict(arguments or {})), sort_keys=True, separators=(",", ":")) + + @classmethod + def _canonicalize(cls, value: Any) -> Any: + """Recursively encode a value with structural type tags.""" + if value is None: + return ["null"] + if isinstance(value, bool): + return ["bool", value] + if isinstance(value, int): + return ["int", value] + if isinstance(value, float): + return ["float", repr(value)] + if isinstance(value, str): + return ["str", value] + if isinstance(value, Mapping): + items = [[cls._canonicalize(key), cls._canonicalize(item)] for key, item in value.items()] + return ["map", sorted(items, key=lambda pair: json.dumps(pair, sort_keys=True))] + if isinstance(value, (list, tuple)): + return ["seq", [cls._canonicalize(item) for item in value]] + value_type = type(value) + return ["obj", f"{value_type.__module__}.{value_type.__qualname__}", str(value)] + + @classmethod + def canonical_args_digest(cls, arguments: Mapping[str, Any] | None) -> str: + """Compute an order-independent digest of the call arguments.""" + rendered = cls.render_canonical_arguments(arguments) + return hashlib.sha256(rendered.encode("utf-8")).hexdigest() + + def grant_approval(self, decision: FunctionAuthorizationDecision, ttl_seconds: float | None = None) -> None: + """Grant a single-use approval for the exact call recorded in the decision.""" + self.approvals.grant( + decision.binding, ttl_seconds if ttl_seconds is not None else self.policy.approval_ttl_seconds + ) + + async def __call__(self, context: "AutoFunctionInvocationContext", next) -> None: + """Authorize the proposed function call before it is dispatched.""" + registered_filters = getattr(context.kernel, "auto_function_invocation_filters", None) + if registered_filters and registered_filters[0][1] is not self and not self._ordering_warned: + self._ordering_warned = True + logger.warning( + "FunctionAuthorizationFilter is not the innermost AUTO_FUNCTION_INVOCATION filter: " + "filters registered after it run between the authorization check and dispatch and " + "could alter the authorized call. Register this filter last to close that gap." + ) + function_name = context.function.fully_qualified_name + arguments = dict(context.arguments or {}) + try: + rendered_arguments = self.render_canonical_arguments(arguments) + except Exception: + # Arguments that cannot be canonicalized (circular references, + # hostile __str__ implementations, ...) fail closed: the call is + # denied and audited rather than dispatched or crashing the gate. + decision = FunctionAuthorizationDecision( + function_name=function_name, + args_digest="uncanonicalizable", + principal=self.policy.principal, + policy_digest=self.policy.policy_digest, + risk=FunctionRiskLevel.CRITICAL, + action=FunctionAuthorizationAction.DENY, + status=FunctionAuthorizationStatus.DENIED, + reason="arguments could not be canonicalized, failing closed", + authority_source="policy", + ) + self.audit_log.append(decision) + logger.warning("Function '%s' denied: %s", function_name, decision.reason) + context.function_result = FunctionResult( + function=context.function.metadata, + value={ + "authorization": "denied", + "decision_id": decision.decision_id, + "function": function_name, + "reason": decision.reason, + "message": ( + f"The call to '{function_name}' was blocked by the function authorization " + "policy and was not executed." + ), + }, + ) + return + args_digest = hashlib.sha256(rendered_arguments.encode("utf-8")).hexdigest() + risk, reason, requires_approval = self.policy.resolve_risk( + function_name, + context.function.plugin_name, + context.function.metadata.additional_properties, + rendered_arguments, + ) + decision = FunctionAuthorizationDecision( + function_name=function_name, + args_digest=args_digest, + principal=self.policy.principal, + policy_digest=self.policy.policy_digest, + risk=risk, + action=self.policy.action_for(risk, requires_approval), + status=FunctionAuthorizationStatus.PENDING_APPROVAL, + reason=reason, + authority_source="policy", + ) + self.audit_log.append(decision) + + if decision.action != FunctionAuthorizationAction.ALLOW: + approval = self.approvals.consume(decision.binding) + if approval == "ok" and decision.action == FunctionAuthorizationAction.REQUIRE_APPROVAL: + decision.action = FunctionAuthorizationAction.ALLOW + decision.authority_source = "user_approval" + decision.status = FunctionAuthorizationStatus.APPROVED + decision.reason = f"approved by '{self.policy.principal}' for this exact call ({reason})" + elif approval == "expired": + decision.status = FunctionAuthorizationStatus.EXPIRED + decision.reason = f"approval expired before dispatch ({reason})" + + if decision.action == FunctionAuthorizationAction.ALLOW: + if decision.status != FunctionAuthorizationStatus.APPROVED: + decision.status = FunctionAuthorizationStatus.APPROVED + logger.debug("Function '%s' authorized: %s", function_name, decision.reason) + try: + await next(context) + except Exception: + decision.status = FunctionAuthorizationStatus.FAILED + raise + decision.status = FunctionAuthorizationStatus.EXECUTED + return + + if decision.action == FunctionAuthorizationAction.DENY: + decision.status = FunctionAuthorizationStatus.DENIED + logger.warning("Function '%s' denied: %s", function_name, decision.reason) + context.function_result = FunctionResult( + function=context.function.metadata, + value={ + "authorization": "denied", + "decision_id": decision.decision_id, + "function": function_name, + "reason": decision.reason, + "message": ( + f"The call to '{function_name}' was blocked by the function authorization " + "policy and was not executed." + ), + }, + ) + return + + if decision.status != FunctionAuthorizationStatus.EXPIRED: + decision.status = FunctionAuthorizationStatus.PENDING_APPROVAL + logger.warning("Function '%s' requires approval: %s", function_name, decision.reason) + context.function_result = FunctionResult( + function=context.function.metadata, + value={ + "authorization": "pending_approval", + "decision_id": decision.decision_id, + "function": function_name, + "args_digest": decision.args_digest, + "reason": decision.reason, + "message": ( + f"The call to '{function_name}' was suspended pending approval and was not " + "executed. A caller with authority can grant the approval and re-issue the " + "identical call." + ), + }, + ) + if self.policy.terminate_on_pending: + context.terminate = True + + +__all__ = [ + "REQUIRES_APPROVAL_PROPERTY", + "RISK_LEVEL_PROPERTY", + "FunctionApprovalStore", + "FunctionAuthorizationAction", + "FunctionAuthorizationDecision", + "FunctionAuthorizationFilter", + "FunctionAuthorizationPolicy", + "FunctionAuthorizationStatus", + "FunctionRiskLevel", +] diff --git a/python/tests/unit/filters/test_function_authorization.py b/python/tests/unit/filters/test_function_authorization.py new file mode 100644 index 000000000000..bec385911595 --- /dev/null +++ b/python/tests/unit/filters/test_function_authorization.py @@ -0,0 +1,471 @@ +# Copyright (c) Microsoft. All rights reserved. + +import time + +from pytest import fixture, raises + +from semantic_kernel import Kernel +from semantic_kernel.contents import ChatHistory +from semantic_kernel.contents.function_call_content import FunctionCallContent +from semantic_kernel.filters import ( + FilterTypes, + FunctionAuthorizationAction, + FunctionAuthorizationFilter, + FunctionAuthorizationPolicy, + FunctionAuthorizationStatus, + FunctionRiskLevel, +) +from semantic_kernel.filters.auto_function_invocation.function_authorization_filter import ( + REQUIRES_APPROVAL_PROPERTY, + RISK_LEVEL_PROPERTY, +) +from semantic_kernel.functions.kernel_function_decorator import kernel_function + + +class BankPlugin: + """A plugin with a destructive function, used to prove calls are (not) dispatched.""" + + def __init__(self): + self.transfers: list[int] = [] + + @kernel_function(name="transfer", description="Transfer an amount of money.") + def transfer(self, amount: int) -> str: + self.transfers.append(amount) + return f"transferred {amount}" + + @kernel_function(name="balance", description="Read the account balance.") + def balance(self) -> str: + return "balance is 100" + + +@fixture +def bank() -> BankPlugin: + return BankPlugin() + + +@fixture +def kernel_with_bank(kernel: Kernel, bank: BankPlugin) -> Kernel: + kernel.add_plugin(bank, plugin_name="bank") + return kernel + + +def transfer_call(amount: int = 10, call_id: str = "call_1") -> FunctionCallContent: + return FunctionCallContent( + id=call_id, plugin_name="bank", function_name="transfer", arguments=f'{{"amount": {amount}}}' + ) + + +def balance_call(call_id: str = "call_2") -> FunctionCallContent: + return FunctionCallContent(id=call_id, plugin_name="bank", function_name="balance", arguments="{}") + + +async def invoke(kernel: Kernel, call: FunctionCallContent, history: ChatHistory | None = None): + history = history if history is not None else ChatHistory() + context = await kernel.invoke_function_call(function_call=call, chat_history=history) + return context, history + + +def add_auth_filter(kernel: Kernel, policy: FunctionAuthorizationPolicy) -> FunctionAuthorizationFilter: + auth_filter = FunctionAuthorizationFilter(policy=policy) + kernel.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, auth_filter) + return auth_filter + + +class TestFailClosedDefaults: + async def test_unclassified_function_requires_approval(self, kernel_with_bank: Kernel, bank: BankPlugin): + """A function with no declared risk fails closed: suspended, never executed.""" + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + _, history = await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + decision = auth_filter.audit_log[-1] + assert decision.action == FunctionAuthorizationAction.REQUIRE_APPROVAL + assert decision.status == FunctionAuthorizationStatus.PENDING_APPROVAL + assert "pending_approval" in str(history.messages[-1].items[0].result) + + async def test_injection_cannot_bypass_the_checkpoint(self, kernel_with_bank: Kernel, bank: BankPlugin): + """Prompt injection can propose a call, but the filter still gates dispatch.""" + add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + malicious = transfer_call(amount=10000) + await invoke(kernel_with_bank, malicious) + + assert bank.transfers == [] + + async def test_invalid_declared_risk_fails_closed(self, kernel_with_bank: Kernel, bank: BankPlugin): + """A typo'd risk level in metadata is treated as the fail-closed default.""" + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + function = kernel_with_bank.get_function("bank", "transfer") + function.metadata.additional_properties = {RISK_LEVEL_PROPERTY: "no-such-level"} + + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + assert auth_filter.audit_log[-1].risk == FunctionRiskLevel.HIGH + + async def test_unmapped_risk_level_denies(self, kernel_with_bank: Kernel, bank: BankPlugin): + """A risk level missing from the action map fails closed to DENY.""" + policy = FunctionAuthorizationPolicy( + risk_overrides={"bank-transfer": FunctionRiskLevel.HIGH}, + action_map={FunctionRiskLevel.LOW: FunctionAuthorizationAction.ALLOW}, + ) + auth_filter = add_auth_filter(kernel_with_bank, policy) + + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + assert auth_filter.audit_log[-1].action == FunctionAuthorizationAction.DENY + + +class TestPolicyClassification: + async def test_low_risk_function_executes(self, kernel_with_bank: Kernel, bank: BankPlugin): + policy = FunctionAuthorizationPolicy(risk_overrides={"bank-balance": FunctionRiskLevel.LOW}) + auth_filter = add_auth_filter(kernel_with_bank, policy) + + _, history = await invoke(kernel_with_bank, balance_call()) + + decision = auth_filter.audit_log[-1] + assert decision.status == FunctionAuthorizationStatus.EXECUTED + assert "balance is 100" in str(history.messages[-1].items[0].result) + + async def test_plugin_wildcard_override(self, kernel_with_bank: Kernel, bank: BankPlugin): + policy = FunctionAuthorizationPolicy(risk_overrides={"bank-*": FunctionRiskLevel.LOW}) + auth_filter = add_auth_filter(kernel_with_bank, policy) + + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [10] + assert auth_filter.audit_log[-1].status == FunctionAuthorizationStatus.EXECUTED + + async def test_exact_override_beats_wildcard(self, kernel_with_bank: Kernel, bank: BankPlugin): + policy = FunctionAuthorizationPolicy( + risk_overrides={"bank-*": FunctionRiskLevel.LOW, "bank-transfer": FunctionRiskLevel.CRITICAL} + ) + auth_filter = add_auth_filter(kernel_with_bank, policy) + + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + assert auth_filter.audit_log[-1].action == FunctionAuthorizationAction.DENY + + async def test_declared_metadata_risk_is_honored(self, kernel_with_bank: Kernel, bank: BankPlugin): + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + function = kernel_with_bank.get_function("bank", "transfer") + function.metadata.additional_properties = {RISK_LEVEL_PROPERTY: "low"} + + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [10] + assert auth_filter.audit_log[-1].risk == FunctionRiskLevel.LOW + + async def test_requires_approval_metadata_overrides_allow(self, kernel_with_bank: Kernel, bank: BankPlugin): + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + function = kernel_with_bank.get_function("bank", "transfer") + function.metadata.additional_properties = { + RISK_LEVEL_PROPERTY: "low", + REQUIRES_APPROVAL_PROPERTY: True, + } + + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + assert auth_filter.audit_log[-1].action == FunctionAuthorizationAction.REQUIRE_APPROVAL + + +class TestKeywordGuard: + async def test_keyword_guard_escalates_low_risk(self, kernel_with_bank: Kernel, bank: BankPlugin): + """The deterministic guard escalates on suspicious argument content; strictest wins.""" + policy = FunctionAuthorizationPolicy( + risk_overrides={"bank-*": FunctionRiskLevel.LOW}, + keyword_guard={"10000": FunctionRiskLevel.CRITICAL}, + ) + auth_filter = add_auth_filter(kernel_with_bank, policy) + + await invoke(kernel_with_bank, transfer_call(amount=10000)) + + assert bank.transfers == [] + decision = auth_filter.audit_log[-1] + assert decision.action == FunctionAuthorizationAction.DENY + assert "keyword guard" in decision.reason + + async def test_keyword_guard_never_lowers_risk(self, kernel_with_bank: Kernel, bank: BankPlugin): + policy = FunctionAuthorizationPolicy( + risk_overrides={"bank-transfer": FunctionRiskLevel.CRITICAL}, + keyword_guard={"amount": FunctionRiskLevel.LOW}, + ) + auth_filter = add_auth_filter(kernel_with_bank, policy) + + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + assert auth_filter.audit_log[-1].risk == FunctionRiskLevel.CRITICAL + + +class TestApprovalBinding: + async def test_grant_then_reissue_executes_once(self, kernel_with_bank: Kernel, bank: BankPlugin): + """Terminate-and-resume: grant the pending decision, re-issue the identical call.""" + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + await invoke(kernel_with_bank, transfer_call()) + pending = auth_filter.audit_log[-1] + assert pending.status == FunctionAuthorizationStatus.PENDING_APPROVAL + + auth_filter.grant_approval(pending) + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [10] + approved = auth_filter.audit_log[-1] + assert approved.status == FunctionAuthorizationStatus.EXECUTED + assert approved.authority_source == "user_approval" + + async def test_changed_arguments_invalidate_approval(self, kernel_with_bank: Kernel, bank: BankPlugin): + """An approval for transfer(10) must not be replayable for transfer(10000).""" + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + await invoke(kernel_with_bank, transfer_call(amount=10)) + auth_filter.grant_approval(auth_filter.audit_log[-1]) + + await invoke(kernel_with_bank, transfer_call(amount=10000)) + + assert bank.transfers == [] + assert auth_filter.audit_log[-1].status == FunctionAuthorizationStatus.PENDING_APPROVAL + + async def test_approval_is_single_use(self, kernel_with_bank: Kernel, bank: BankPlugin): + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + await invoke(kernel_with_bank, transfer_call()) + auth_filter.grant_approval(auth_filter.audit_log[-1]) + + await invoke(kernel_with_bank, transfer_call()) + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [10] + assert auth_filter.audit_log[-1].status == FunctionAuthorizationStatus.PENDING_APPROVAL + + async def test_expired_approval_is_terminal_non_execution(self, kernel_with_bank: Kernel, bank: BankPlugin): + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + await invoke(kernel_with_bank, transfer_call()) + pending = auth_filter.audit_log[-1] + auth_filter.grant_approval(pending) + auth_filter.approvals._approvals[pending.binding] = time.time() - 1.0 + + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + assert auth_filter.audit_log[-1].status == FunctionAuthorizationStatus.EXPIRED + + async def test_non_finite_or_non_positive_ttl_is_rejected(self, kernel_with_bank: Kernel, bank: BankPlugin): + """NaN/inf/zero TTLs would create approvals that never expire; they are rejected.""" + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + await invoke(kernel_with_bank, transfer_call()) + pending = auth_filter.audit_log[-1] + + for bad_ttl in (float("nan"), float("inf"), 0.0, -1.0): + with raises(ValueError): + auth_filter.grant_approval(pending, ttl_seconds=bad_ttl) + + assert bank.transfers == [] + + async def test_policy_change_invalidates_approval(self, kernel_with_bank: Kernel, bank: BankPlugin): + """An approval is bound to the policy snapshot that produced it.""" + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + await invoke(kernel_with_bank, transfer_call()) + auth_filter.grant_approval(auth_filter.audit_log[-1]) + + auth_filter.policy = FunctionAuthorizationPolicy(principal="someone_else") + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + + async def test_denied_cannot_be_converted_by_approval(self, kernel_with_bank: Kernel, bank: BankPlugin): + """A DENY decision is terminal: retrying with a granted approval stays denied.""" + policy = FunctionAuthorizationPolicy(risk_overrides={"bank-transfer": FunctionRiskLevel.CRITICAL}) + auth_filter = add_auth_filter(kernel_with_bank, policy) + + await invoke(kernel_with_bank, transfer_call()) + denied = auth_filter.audit_log[-1] + assert denied.status == FunctionAuthorizationStatus.DENIED + + auth_filter.grant_approval(denied) + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + assert auth_filter.audit_log[-1].status == FunctionAuthorizationStatus.DENIED + + +class TestLoopAndModelSignals: + async def test_pending_terminates_loop_by_default(self, kernel_with_bank: Kernel, bank: BankPlugin): + add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + context, _ = await invoke(kernel_with_bank, transfer_call()) + + assert context is not None + assert context.terminate is True + + async def test_pending_without_terminate(self, kernel_with_bank: Kernel, bank: BankPlugin): + add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy(terminate_on_pending=False)) + + context, _ = await invoke(kernel_with_bank, transfer_call()) + + assert context is None + + async def test_deny_feeds_refusal_to_model_without_terminating(self, kernel_with_bank: Kernel, bank: BankPlugin): + """The model gets a clean structured 'blocked' signal, not a silent no-op.""" + policy = FunctionAuthorizationPolicy(risk_overrides={"bank-transfer": FunctionRiskLevel.CRITICAL}) + add_auth_filter(kernel_with_bank, policy) + + context, history = await invoke(kernel_with_bank, transfer_call()) + + assert context is None + result = str(history.messages[-1].items[0].result) + assert "denied" in result + assert "not executed" in result + + +class TestAuditLog: + async def test_audit_distinguishes_lifecycle_states(self, kernel_with_bank: Kernel, bank: BankPlugin): + """Proposed→pending, approved→executed, denied and expired are distinct records.""" + auth_filter = add_auth_filter(kernel_with_bank, FunctionAuthorizationPolicy()) + + await invoke(kernel_with_bank, transfer_call()) + auth_filter.grant_approval(auth_filter.audit_log[-1]) + await invoke(kernel_with_bank, transfer_call()) + await invoke(kernel_with_bank, transfer_call()) + pending = auth_filter.audit_log[-1] + auth_filter.grant_approval(pending) + auth_filter.approvals._approvals[pending.binding] = time.time() - 1.0 + await invoke(kernel_with_bank, transfer_call()) + + states = [decision.status for decision in auth_filter.audit_log] + assert states == [ + FunctionAuthorizationStatus.PENDING_APPROVAL, + FunctionAuthorizationStatus.EXECUTED, + FunctionAuthorizationStatus.PENDING_APPROVAL, + FunctionAuthorizationStatus.EXPIRED, + ] + assert len({decision.decision_id for decision in auth_filter.audit_log}) == 4 + + def test_args_digest_is_order_independent(self): + digest_a = FunctionAuthorizationFilter.canonical_args_digest({"a": 1, "b": 2}) + digest_b = FunctionAuthorizationFilter.canonical_args_digest({"b": 2, "a": 1}) + digest_c = FunctionAuthorizationFilter.canonical_args_digest({"a": 1, "b": 3}) + + assert digest_a == digest_b + assert digest_a != digest_c + + def test_decision_binding_covers_all_dimensions(self): + base = dict( + function_name="bank-transfer", + args_digest="d1", + principal="p1", + policy_digest="s1", + risk=FunctionRiskLevel.HIGH, + action=FunctionAuthorizationAction.REQUIRE_APPROVAL, + status=FunctionAuthorizationStatus.PENDING_APPROVAL, + reason="r", + authority_source="policy", + ) + from semantic_kernel.filters import FunctionAuthorizationDecision + + reference = FunctionAuthorizationDecision(**base) + for field, changed in [ + ("function_name", "bank-balance"), + ("args_digest", "d2"), + ("principal", "p2"), + ("policy_digest", "s2"), + ]: + variant = FunctionAuthorizationDecision(**{**base, field: changed}) + assert variant.binding != reference.binding, field + + def test_digest_distinguishes_type_from_lookalike_str(self): + """An object whose __str__ mimics an approved value must not collide.""" + + class Lookalike: + def __str__(self): + return "approved" + + digest_str = FunctionAuthorizationFilter.canonical_args_digest({"x": "approved"}) + digest_obj = FunctionAuthorizationFilter.canonical_args_digest({"x": Lookalike()}) + + assert digest_str != digest_obj + + # A native string crafted to mimic the object's tagged rendering must + # not collide either: the structural type tag keeps the classes apart. + rendered_obj = FunctionAuthorizationFilter.render_canonical_arguments({"x": Lookalike()}) + mimic = rendered_obj.split('["obj","', 1)[-1] + digest_mimic = FunctionAuthorizationFilter.canonical_args_digest({"x": mimic}) + assert digest_mimic != digest_obj + + async def test_uncanonicalizable_arguments_fail_closed(self, kernel_with_bank: Kernel, bank: BankPlugin): + """Circular arguments deny and audit instead of crashing the gate open.""" + policy = FunctionAuthorizationPolicy(risk_overrides={"bank-*": FunctionRiskLevel.LOW}) + auth_filter = add_auth_filter(kernel_with_bank, policy) + cyclic: dict = {} + cyclic["self"] = cyclic + + async def poison_arguments(context, next): + context.arguments["amount"] = cyclic + await next(context) + + kernel_with_bank.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, poison_arguments) + # poison_arguments was added last, so it runs innermost; re-add the + # authorization filter so it is innermost again, after the poisoning. + kernel_with_bank.remove_filter(filter_id=id(auth_filter)) + kernel_with_bank.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, auth_filter) + + _, history = await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + decision = auth_filter.audit_log[-1] + assert decision.status == FunctionAuthorizationStatus.DENIED + assert decision.args_digest == "uncanonicalizable" + assert "denied" in str(history.messages[-1].items[0].result) + + async def test_dispatch_failure_is_recorded_as_failed(self, kernel_with_bank: Kernel, bank: BankPlugin): + """If dispatch raises after authorization, the audit shows FAILED, not EXECUTED.""" + policy = FunctionAuthorizationPolicy(risk_overrides={"bank-*": FunctionRiskLevel.LOW}) + auth_filter = add_auth_filter(kernel_with_bank, policy) + + async def exploding_downstream(context, next): + raise RuntimeError("downstream blew up") + + # Added last -> runs innermost, between the authorization filter and dispatch. + kernel_with_bank.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, exploding_downstream) + + with raises(RuntimeError): + await invoke(kernel_with_bank, transfer_call()) + + assert bank.transfers == [] + assert auth_filter.audit_log[-1].status == FunctionAuthorizationStatus.FAILED + + async def test_warns_when_not_innermost_filter(self, kernel_with_bank: Kernel, bank: BankPlugin, caplog): + """A filter registered between authorization and dispatch triggers a warning.""" + policy = FunctionAuthorizationPolicy(risk_overrides={"bank-*": FunctionRiskLevel.LOW}) + add_auth_filter(kernel_with_bank, policy) + + async def passthrough(context, next): + await next(context) + + # Added last -> runs innermost, between the authorization filter and dispatch. + kernel_with_bank.add_filter(FilterTypes.AUTO_FUNCTION_INVOCATION, passthrough) + + with caplog.at_level("WARNING"): + await invoke(kernel_with_bank, transfer_call()) + + assert any("innermost" in record.message for record in caplog.records) + + def test_approval_store_expiry(self): + from semantic_kernel.filters import FunctionApprovalStore + + store = FunctionApprovalStore() + store.grant("binding", ttl_seconds=100) + assert store.consume("binding") == "ok" + assert store.consume("binding") == "absent" + + store.grant("binding", ttl_seconds=100) + store._approvals["binding"] = time.time() - 1.0 + assert store.consume("binding") == "expired"