diff --git a/docs/VAULTAK_SECURITY.md b/docs/VAULTAK_SECURITY.md new file mode 100644 index 000000000000..77e260ead04a --- /dev/null +++ b/docs/VAULTAK_SECURITY.md @@ -0,0 +1,135 @@ +# Vaultak Runtime Security Integration + +[Vaultak](https://vaultak.com) is a runtime security platform that wraps Semantic Kernel agents +with real-time protection. Every plugin function call and LLM tool selection is risk-scored (0–10), +checked against your policy rules, and automatically blocked before it reaches your production +systems — without changing your agent logic. + +## Why use Vaultak with Semantic Kernel? + +SK agents given plugins can cause real damage: deleted records, leaked PII, unauthorized API calls. +Vaultak adds a security layer via SK's native **filter** system: + +- **Risk scoring** on every plugin function call before it executes +- **Policy enforcement** — block calls that violate your rules +- **PII masking** — strip sensitive data from plugin outputs +- **Audit trail** — every call, every score, in your Vaultak dashboard + +## How it works + +Semantic Kernel's [filter system](https://learn.microsoft.com/en-us/semantic-kernel/concepts/enterprise-readiness/filters) +is the native hook for security and observability. Vaultak registers two filters: + +| Filter type | When it runs | Vaultak action | +|---|---|---| +| `FunctionInvocationFilter` | Every `kernel.invoke()` call (plugin functions only) | Risk-scores the call; raises `OperationCancelledException` if score meets or exceeds threshold; masks PII in output | +| `AutoFunctionInvocationFilter` | Each LLM-selected tool call (auto function calling) | Risk-scores the call; sets `context.terminate = True` to stop the loop if above threshold | + +## Installation + +```bash +pip install vaultak semantic-kernel +``` + +Sign up at [vaultak.com](https://vaultak.com) to get your API key (starts with `vtk_`). + +## Quick start + +```python +import asyncio +import os +from collections.abc import Callable, Coroutine +from typing import Any + +from semantic_kernel import Kernel +from semantic_kernel.connectors.ai import FunctionChoiceBehavior +from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings +from semantic_kernel.core_plugins import MathPlugin, TimePlugin +from semantic_kernel.exceptions import OperationCancelledException +from semantic_kernel.filters import AutoFunctionInvocationContext, FilterTypes, FunctionInvocationContext +from semantic_kernel.functions import FunctionResult, KernelArguments + +from vaultak import Vaultak + +_api_key = os.environ.get("VAULTAK_API_KEY") +if not _api_key: + raise ValueError( + "VAULTAK_API_KEY environment variable is not set. " + "Sign up at https://vaultak.com to get your API key." + ) + +RISK_THRESHOLD = 7.0 +vt = Vaultak(api_key=_api_key, agent_name="sk-agent") + +kernel = Kernel() +kernel.add_service(OpenAIChatCompletion(service_id="chat")) +kernel.add_plugin(MathPlugin(), plugin_name="math") +kernel.add_plugin(TimePlugin(), plugin_name="time") + + +@kernel.filter(FilterTypes.FUNCTION_INVOCATION) +async def vaultak_function_filter( + context: FunctionInvocationContext, + next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]], +) -> None: + # Skip inline prompt functions — only score real plugin calls + if context.function.plugin_name is None: + await next(context) + return + action = f"{context.function.plugin_name}-{context.function.name}" + result = await asyncio.to_thread(vt.score_action, action=action, context=dict(context.arguments or {})) + if result.score >= RISK_THRESHOLD: + raise OperationCancelledException( + f"[Vaultak] '{action}' blocked — risk {result.score:.1f}/10 " + f"meets or exceeds threshold {RISK_THRESHOLD}. Review at app.vaultak.com" + ) + await asyncio.to_thread(vt.check_policy, tool_name=action, input_data=str(context.arguments)) + await next(context) + if context.result and context.result.value: + masked = await asyncio.to_thread(vt.mask_pii, str(context.result.value)) + context.result = FunctionResult(function=context.function, value=masked) + + +@kernel.filter(FilterTypes.AUTO_FUNCTION_INVOCATION) +async def vaultak_auto_filter( + context: AutoFunctionInvocationContext, + next: Callable[[AutoFunctionInvocationContext], Coroutine[Any, Any, None]], +) -> None: + action = f"{context.function.plugin_name or 'kernel'}-{context.function.name}" + result = await asyncio.to_thread(vt.score_action, action=action, context={"auto_invoke": "true"}) + if result.score >= RISK_THRESHOLD: + context.terminate = True # Stop auto-invocation loop cleanly + return + await next(context) +``` + +## Configuration + +| Parameter | Type | Default | Description | +|---|---|---|---| +| `api_key` | `str` | — | Your Vaultak API key — required | +| `agent_name` | `str` | `"sk-agent"` | Label for this kernel in the Vaultak dashboard | +| `risk_threshold` | `float` | `7.0` | Score (0–10) above which calls are blocked | +| `verbose` | `bool` | `False` | Log every scored action to stdout | + +## What gets monitored + +| SK event | Vaultak action | +|---|---| +| `kernel.invoke()` (explicit call) | Risk-scores the plugin + function; blocks if above threshold | +| Auto function call selected by LLM | Risk-scores; terminates auto-invocation loop if above threshold | +| Plugin output returned | Scans for PII and masks before the result propagates | +| Policy check fails | Raises exception with dashboard URL | + +## Complete sample + +See [`python/samples/concepts/filtering/vaultak_security_filter.py`](../python/samples/concepts/filtering/vaultak_security_filter.py) +for a runnable end-to-end example with MathPlugin and TimePlugin. + +## Links + +- [Vaultak documentation](https://docs.vaultak.com) +- [PyPI: `vaultak`](https://pypi.org/project/vaultak) +- [GitHub](https://github.com/vaultak/vaultak-python) +- [Dashboard](https://app.vaultak.com) +- [SK Filters concept](https://learn.microsoft.com/en-us/semantic-kernel/concepts/enterprise-readiness/filters) diff --git a/python/samples/concepts/filtering/vaultak_security_filter.py b/python/samples/concepts/filtering/vaultak_security_filter.py new file mode 100644 index 000000000000..f6db9df30b22 --- /dev/null +++ b/python/samples/concepts/filtering/vaultak_security_filter.py @@ -0,0 +1,165 @@ +# Copyright (c) Microsoft. All rights reserved. + +""" +Vaultak runtime security filter for Semantic Kernel. + +Demonstrates how to integrate Vaultak (https://vaultak.com) with Semantic Kernel +using two filter types: + +- FunctionInvocationFilter — risk-scores every plugin function call before it + executes and raises OperationCancelledException when the score meets or + exceeds the threshold. +- AutoFunctionInvocationFilter — intercepts each tool call chosen by the LLM + during auto-function-calling, allowing per-call blocking without stopping + the whole invocation loop. + +Install: + pip install vaultak semantic-kernel + +Set environment variables: + OPENAI_API_KEY=sk-... + VAULTAK_API_KEY=vtk_... +""" + +import asyncio +import os +from collections.abc import Callable, Coroutine +from typing import Any + +from semantic_kernel import Kernel +from semantic_kernel.connectors.ai import FunctionChoiceBehavior +from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings +from semantic_kernel.contents import ChatHistory +from semantic_kernel.core_plugins import MathPlugin, TimePlugin +from semantic_kernel.exceptions import OperationCancelledException +from semantic_kernel.filters import AutoFunctionInvocationContext, FilterTypes, FunctionInvocationContext +from semantic_kernel.functions import FunctionResult, KernelArguments + +from vaultak import Vaultak + +_api_key = os.environ.get("VAULTAK_API_KEY") +if not _api_key: + raise ValueError( + "VAULTAK_API_KEY environment variable is not set. " + "Sign up at https://vaultak.com to get your API key." + ) + +RISK_THRESHOLD = 7.0 # Block function calls with a risk score that meets or exceeds this value + +vt = Vaultak(api_key=_api_key, agent_name="sk-agent") + +kernel = Kernel() +kernel.add_service(OpenAIChatCompletion(service_id="chat")) +kernel.add_plugin(MathPlugin(), plugin_name="math") +kernel.add_plugin(TimePlugin(), plugin_name="time") + + +# --------------------------------------------------------------------------- +# Filter 1: FunctionInvocationFilter +# Runs around every KernelFunction invocation (including prompt functions). +# Use this to risk-score explicit kernel.invoke() calls. +# --------------------------------------------------------------------------- +@kernel.filter(FilterTypes.FUNCTION_INVOCATION) +async def vaultak_function_filter( + context: FunctionInvocationContext, + next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]], +) -> None: + """Risk-score every kernel function call before executing it.""" + # Skip prompt functions (plugin_name is None for inline prompts) + if context.function.plugin_name is None: + await next(context) + return + + plugin_name = context.function.plugin_name + function_name = context.function.name + action = f"{plugin_name}-{function_name}" + + # Collect function arguments as context for the risk scorer + args_context = {k: str(v) for k, v in (context.arguments or {}).items()} + + # Wrap synchronous SDK calls with asyncio.to_thread to avoid blocking the event loop + result = await asyncio.to_thread(vt.score_action, action=action, context=args_context) + + if result.score >= RISK_THRESHOLD: + raise OperationCancelledException( + f"[Vaultak] Function '{action}' blocked — risk score {result.score:.1f}/10 " + f"meets or exceeds threshold {RISK_THRESHOLD}. Review at app.vaultak.com" + ) + + # Check against configured policy rules + await asyncio.to_thread(vt.check_policy, tool_name=action, input_data=str(args_context)) + + await next(context) + + # Scan the function output for PII before it propagates + if context.result and context.result.value: + raw_output = str(context.result.value) + masked = await asyncio.to_thread(vt.mask_pii, raw_output) + context.result = FunctionResult(function=context.function, value=masked) + + +# --------------------------------------------------------------------------- +# Filter 2: AutoFunctionInvocationFilter +# Runs for each tool call the LLM selects during auto-function-calling. +# Use this to risk-score or block individual tool calls mid-conversation. +# --------------------------------------------------------------------------- +@kernel.filter(FilterTypes.AUTO_FUNCTION_INVOCATION) +async def vaultak_auto_filter( + context: AutoFunctionInvocationContext, + next: Callable[[AutoFunctionInvocationContext], Coroutine[Any, Any, None]], +) -> None: + """Risk-score each LLM-selected tool call during auto function invocation.""" + plugin_name = context.function.plugin_name or "kernel" + function_name = context.function.name + action = f"{plugin_name}-{function_name}" + + result = await asyncio.to_thread(vt.score_action, action=action, context={"auto_invoke": "true"}) + + if result.score >= RISK_THRESHOLD: + # Setting terminate=True stops the auto-invocation loop cleanly + # instead of raising an exception that would surface to the user. + context.terminate = True + return + + await asyncio.to_thread(vt.check_policy, tool_name=action, input_data=action) + await next(context) + + +# --------------------------------------------------------------------------- +# Demo: run a short agent conversation +# --------------------------------------------------------------------------- +execution_settings = OpenAIChatPromptExecutionSettings( + service_id="chat", + max_tokens=1000, + function_choice_behavior=FunctionChoiceBehavior.Auto( + filters={"included_plugins": ["math", "time"]} + ), +) + +history = ChatHistory() +history.add_system_message("You are a helpful assistant. Use math and time tools when asked.") + + +async def chat(user_input: str) -> None: + history.add_user_message(user_input) + arguments = KernelArguments( + settings=execution_settings, + chat_history=history, + user_input=user_input, + ) + result = await kernel.invoke_prompt( + "{{$chat_history}}{{$user_input}}", + arguments=arguments, + ) + print(f"Assistant: {result}") + history.add_assistant_message(str(result)) + + +async def main() -> None: + print("Vaultak security filters active. Every tool call is risk-scored before execution.") + await chat("What is 42 multiplied by 7?") + await chat("What time is it right now?") + + +if __name__ == "__main__": + asyncio.run(main())