Python: inversed the filter logic for InMemory vector stores (#13457)

### Motivation and Context

<!-- Thank you for your contribution to the semantic-kernel repo!
Please help reviewers and future users, providing the following
information:
  1. Why is this change required?
  2. What problem does it solve?
  3. What scenario does it contribute to?
  4. If it fixes an open issue, please link to the issue here.
-->
Fixes some issues with the forbidden list approach for the in-memory
vector store filtering.

### Description

<!-- Describe your changes, the overall approach, the underlying design.
These notes will help understanding how your code works. Thanks! -->

### Contribution Checklist

<!-- Before submitting this PR, please make sure: -->

- [x] The code builds clean without any errors or warnings
- [x] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [x] All unit tests pass, and I have added new tests where possible
- [ ] I didn't break anyone 😄

Author: eavanvalkenburg

python/semantic_kernel/connectors/in_memory.py

@@ -92,6 +92,80 @@ class InMemoryCollection(
     supported_key_types: ClassVar[set[str] | None] = {"str", "int", "float"}
     supported_search_types: ClassVar[set[SearchType]] = {SearchType.VECTOR}
 
+    # Allowlist of AST node types permitted in filter expressions.
+    # This can be overridden in subclasses to extend or restrict allowed operations.
+    allowed_filter_ast_nodes: ClassVar[set[type]] = {
+        ast.Expression,
+        ast.Lambda,
+        ast.arguments,
+        ast.arg,
+        # Comparisons and boolean operations
+        ast.Compare,
+        ast.BoolOp,
+        ast.UnaryOp,
+        ast.And,
+        ast.Or,
+        ast.Not,
+        ast.Eq,
+        ast.NotEq,
+        ast.Lt,
+        ast.LtE,
+        ast.Gt,
+        ast.GtE,
+        ast.In,
+        ast.NotIn,
+        ast.Is,
+        ast.IsNot,
+        # Data access
+        ast.Name,
+        ast.Load,
+        ast.Attribute,
+        ast.Subscript,
+        ast.Index,  # For Python 3.8 compatibility
+        ast.Slice,
+        # Literals
+        ast.Constant,
+        ast.List,
+        ast.Tuple,
+        ast.Set,
+        ast.Dict,
+        # Basic arithmetic (useful for computed comparisons)
+        ast.BinOp,
+        ast.Add,
+        ast.Sub,
+        ast.Mult,
+        ast.Div,
+        ast.Mod,
+        ast.FloorDiv,
+        # Function calls (restricted to safe builtins separately)
+        ast.Call,
+    }
+
+    # Allowlist of function/method names that can be called in filter expressions.
+    allowed_filter_functions: ClassVar[set[str]] = {
+        "len",
+        "str",
+        "int",
+        "float",
+        "bool",
+        "abs",
+        "min",
+        "max",
+        "sum",
+        "any",
+        "all",
+        "lower",
+        "upper",
+        "strip",
+        "startswith",
+        "endswith",
+        "contains",
+        "get",
+        "keys",
+        "values",
+        "items",
+    }
+
     def __init__(
         self,
         record_type: type[TModel],
@@ -100,7 +174,17 @@ def __init__(
         embedding_generator: EmbeddingGeneratorBase | None = None,
         **kwargs: Any,
     ):
-        """Create a In Memory Collection."""
+        """Create a In Memory Collection.
+
+        In Memory collections are ephemeral and exist only in memory.
+        They do not persist data to disk or any external storage.
+
+        > [Important]
+        > Filters are powerful things, so make sure to not allow untrusted input here.
+        > Filters for this collection are parsed and evaluated using Python's `ast` module, so code might be executed.
+        > We only allow certain AST nodes and functions to be used in the filter expressions to mitigate security risks.
+
+        """
         super().__init__(
             record_type=record_type,
             definition=definition,
@@ -243,39 +327,67 @@ def _get_filtered_records(self, options: VectorSearchOptions) -> dict[TKey, Attr
         return filtered_records
 
     def _parse_and_validate_filter(self, filter_str: str) -> Callable:
-        """Parse and validate a string filter as a lambda expression, then return the callable."""
-        forbidden_names = {
-            "__import__",
-            "open",
-            "eval",
-            "exec",
-            "__builtins__",
-            "__class__",
-            "__bases__",
-            "__subclasses__",
-        }
+        """Parse and validate a string filter as a lambda expression, then return the callable.
+
+        Uses an allowlist approach - only explicitly permitted AST node types and function names
+        are allowed. This can be customized by overriding `allowed_filter_ast_nodes` and
+        `allowed_filter_functions` class attributes.
+        """
         try:
             tree = ast.parse(filter_str, mode="eval")
         except SyntaxError as e:
             raise VectorStoreOperationException(f"Filter string is not valid Python: {e}") from e
-        # Only allow lambda expressions
+
+        # Only allow lambda expressions at the top level
         if not (isinstance(tree, ast.Expression) and isinstance(tree.body, ast.Lambda)):
             raise VectorStoreOperationException(
                 "Filter string must be a lambda expression, e.g. 'lambda x: x.key == 1'"
             )
-        # Walk the AST to look for forbidden names and attribute access
+
+        # Get the lambda parameter name(s) to allow them as valid Name nodes
+        lambda_node = tree.body
+        lambda_param_names = {arg.arg for arg in lambda_node.args.args}
+
+        # Walk the AST to validate all nodes against the allowlist
         for node in ast.walk(tree):
-            if isinstance(node, ast.Name) and node.id in forbidden_names:
-                raise VectorStoreOperationException(f"Use of '{node.id}' is not allowed in filter expressions.")
-            if isinstance(node, ast.Attribute) and node.attr in forbidden_names:
-                raise VectorStoreOperationException(f"Use of '{node.attr}' is not allowed in filter expressions.")
+            node_type = type(node)
+
+            # Check if the node type is allowed
+            if node_type not in self.allowed_filter_ast_nodes:
+                raise VectorStoreOperationException(
+                    f"AST node type '{node_type.__name__}' is not allowed in filter expressions."
+                )
+
+            # For Name nodes, only allow the lambda parameter
+            if isinstance(node, ast.Name) and node.id not in lambda_param_names:
+                raise VectorStoreOperationException(
+                    f"Use of name '{node.id}' is not allowed in filter expressions. "
+                    f"Only the lambda parameter(s) ({', '.join(lambda_param_names)}) can be used."
+                )
+
+            # For Call nodes, validate that only allowed functions are called
+            if isinstance(node, ast.Call):
+                func_name = None
+                if isinstance(node.func, ast.Name):
+                    func_name = node.func.id
+                elif isinstance(node.func, ast.Attribute):
+                    func_name = node.func.attr
+
+                if func_name and func_name not in self.allowed_filter_functions:
+                    raise VectorStoreOperationException(
+                        f"Function '{func_name}' is not allowed in filter expressions. "
+                        f"Allowed functions: {', '.join(sorted(self.allowed_filter_functions))}"
+                    )
+
         try:
             code = compile(tree, filename="<filter>", mode="eval")
             func = eval(code, {"__builtins__": {}}, {})  # nosec
         except Exception as e:
             raise VectorStoreOperationException(f"Error compiling filter: {e}") from e
+
         if not callable(func):
             raise VectorStoreOperationException("Compiled filter is not callable.")
+
         return func
 
     def _run_filter(self, filter: Callable, record: AttributeDict[TAKey, TAValue]) -> bool:

python/tests/unit/core_plugins/test_conversation_summary_plugin_unit.py

@@ -1,15 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from unittest.mock import AsyncMock, Mock
 
-import pytest
-
-from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase
-from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
-from semantic_kernel.contents.chat_message_content import ChatMessageContent
 from semantic_kernel.core_plugins.conversation_summary_plugin import ConversationSummaryPlugin
-from semantic_kernel.functions.kernel_arguments import KernelArguments
-from semantic_kernel.kernel import Kernel
 from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig
 
 
@@ -25,23 +17,3 @@ def test_conversation_summary_plugin_with_deprecated_value(kernel):
     plugin = ConversationSummaryPlugin(config, kernel=kernel)
     assert plugin._summarizeConversationFunction is not None
     assert plugin.return_key == "summary"
-
-
-@pytest.mark.asyncio
-async def test_summarize_conversation(kernel: Kernel):
-    service = AsyncMock(spec=ChatCompletionClientBase)
-    service.service_id = "default"
-    service.get_chat_message_contents = AsyncMock(
-        return_value=[ChatMessageContent(role="assistant", content="Hello World!")]
-    )
-    service.get_prompt_execution_settings_class = Mock(return_value=PromptExecutionSettings)
-    kernel.add_service(service)
-    config = PromptTemplateConfig(
-        name="test", description="test", execution_settings={"default": PromptExecutionSettings()}
-    )
-    kernel.add_plugin(ConversationSummaryPlugin(config), "summarizer")
-    args = KernelArguments(input="Hello World!")
-
-    await kernel.invoke(plugin_name="summarizer", function_name="SummarizeConversation", arguments=args)
-    args["summary"] == "Hello world"
-    service.get_chat_message_contents.assert_called_once()