diff --git a/python/semantic_kernel/processes/dapr_runtime/actors/step_actor.py b/python/semantic_kernel/processes/dapr_runtime/actors/step_actor.py index 5dbb3457d29b..cc2e3c0c23cf 100644 --- a/python/semantic_kernel/processes/dapr_runtime/actors/step_actor.py +++ b/python/semantic_kernel/processes/dapr_runtime/actors/step_actor.py @@ -1,10 +1,9 @@ # Copyright (c) Microsoft. All rights reserved. import asyncio -import importlib import json import logging -from collections.abc import Callable +from collections.abc import Callable, Sequence from inspect import isawaitable from queue import Queue from typing import Any @@ -38,7 +37,11 @@ from semantic_kernel.processes.process_message import ProcessMessage from semantic_kernel.processes.process_message_factory import ProcessMessageFactory from semantic_kernel.processes.process_types import get_generic_state_type -from semantic_kernel.processes.step_utils import find_input_channels, get_fully_qualified_name +from semantic_kernel.processes.step_utils import ( + find_input_channels, + get_fully_qualified_name, + get_step_class_from_qualified_name, +) from semantic_kernel.utils.feature_stage_decorator import experimental logger: logging.Logger = logging.getLogger(__name__) @@ -48,7 +51,14 @@ class StepActor(Actor, StepInterface, KernelProcessMessageChannel): """Represents a step actor that follows the Step abstract class.""" - def __init__(self, ctx: ActorRuntimeContext, actor_id: ActorId, kernel: Kernel, factories: dict[str, Callable]): + def __init__( + self, + ctx: ActorRuntimeContext, + actor_id: ActorId, + kernel: Kernel, + factories: dict[str, Callable], + allowed_module_prefixes: Sequence[str] | None = None, + ): """Initializes a new instance of StepActor. Args: @@ -56,10 +66,14 @@ def __init__(self, ctx: ActorRuntimeContext, actor_id: ActorId, kernel: Kernel, actor_id: The unique ID for the actor. kernel: The Kernel dependency to be injected. factories: The factory dictionary to use for creating the step. + allowed_module_prefixes: Optional sequence of module prefixes that are allowed + for step class loading. If provided, step classes must come from modules + starting with one of these prefixes. """ super().__init__(ctx, actor_id) self.kernel = kernel self.factories: dict[str, Callable] = factories + self.allowed_module_prefixes: Sequence[str] | None = allowed_module_prefixes self.parent_process_id: str | None = None self.step_info: DaprStepInfo | None = None self.initialize_task: bool | None = False @@ -168,12 +182,6 @@ async def process_incoming_messages(self): await self._state_manager.try_add_state(ActorStateKeys.StepIncomingMessagesState.value, messages_to_save) await self._state_manager.save_state() - def _get_class_from_string(self, full_class_name: str): - """Gets a class from a string.""" - module_name, class_name = full_class_name.rsplit(".", 1) - module = importlib.import_module(module_name) - return getattr(module, class_name) - async def activate_step(self): """Initializes the step.""" # Instantiate an instance of the inner step object and retrieve its class reference. @@ -184,7 +192,10 @@ async def activate_step(self): step_cls = step_object.__class__ step_instance: KernelProcessStep = step_object # type: ignore else: - step_cls = self._get_class_from_string(self.inner_step_type) + step_cls = get_step_class_from_qualified_name( + self.inner_step_type, + allowed_module_prefixes=self.allowed_module_prefixes, + ) step_instance: KernelProcessStep = step_cls() # type: ignore kernel_plugin = self.kernel.add_plugin( diff --git a/python/semantic_kernel/processes/dapr_runtime/dapr_process_info.py b/python/semantic_kernel/processes/dapr_runtime/dapr_process_info.py index f573c20c2726..3df00e2d0cd1 100644 --- a/python/semantic_kernel/processes/dapr_runtime/dapr_process_info.py +++ b/python/semantic_kernel/processes/dapr_runtime/dapr_process_info.py @@ -1,7 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. -from collections.abc import MutableSequence +from collections.abc import MutableSequence, Sequence from typing import Literal from pydantic import Field @@ -20,17 +20,23 @@ class DaprProcessInfo(DaprStepInfo): type: Literal["DaprProcessInfo"] = "DaprProcessInfo" # type: ignore steps: MutableSequence["DaprStepInfo | DaprProcessInfo"] = Field(default_factory=list) - def to_kernel_process(self) -> KernelProcess: - """Converts the Dapr process info to a kernel process.""" + def to_kernel_process(self, allowed_module_prefixes: Sequence[str] | None = None) -> KernelProcess: + """Converts the Dapr process info to a kernel process. + + Args: + allowed_module_prefixes: Optional list of module prefixes that are allowed + for step class loading. If provided, step classes must come from modules + starting with one of these prefixes. + """ if not isinstance(self.state, KernelProcessState): raise ValueError("State must be a kernel process state") steps: list[KernelProcessStepInfo] = [] for step in self.steps: if isinstance(step, DaprProcessInfo): - steps.append(step.to_kernel_process()) + steps.append(step.to_kernel_process(allowed_module_prefixes=allowed_module_prefixes)) else: - steps.append(step.to_kernel_process_step_info()) + steps.append(step.to_kernel_process_step_info(allowed_module_prefixes=allowed_module_prefixes)) return KernelProcess(state=self.state, steps=steps, edges=self.edges) diff --git a/python/semantic_kernel/processes/dapr_runtime/dapr_step_info.py b/python/semantic_kernel/processes/dapr_runtime/dapr_step_info.py index 055741c0812e..d34073604939 100644 --- a/python/semantic_kernel/processes/dapr_runtime/dapr_step_info.py +++ b/python/semantic_kernel/processes/dapr_runtime/dapr_step_info.py @@ -1,6 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. -import importlib +from collections.abc import Sequence from typing import Literal from pydantic import Field @@ -11,7 +11,7 @@ from semantic_kernel.processes.kernel_process.kernel_process_state import KernelProcessState from semantic_kernel.processes.kernel_process.kernel_process_step_info import KernelProcessStepInfo from semantic_kernel.processes.kernel_process.kernel_process_step_state import KernelProcessStepState -from semantic_kernel.processes.step_utils import get_fully_qualified_name +from semantic_kernel.processes.step_utils import get_fully_qualified_name, get_step_class_from_qualified_name from semantic_kernel.utils.feature_stage_decorator import experimental @@ -24,13 +24,20 @@ class DaprStepInfo(KernelBaseModel): state: KernelProcessState | KernelProcessStepState edges: dict[str, list[KernelProcessEdge]] = Field(default_factory=dict) - def to_kernel_process_step_info(self) -> KernelProcessStepInfo: - """Converts the Dapr step info to a kernel process step info.""" - inner_step_type = self._get_class_from_string(self.inner_step_python_type) - if inner_step_type is None: - raise KernelException( - f"Unable to create inner step type from assembly qualified name `{self.inner_step_python_type}`" - ) + def to_kernel_process_step_info( + self, allowed_module_prefixes: Sequence[str] | None = None + ) -> KernelProcessStepInfo: + """Converts the Dapr step info to a kernel process step info. + + Args: + allowed_module_prefixes: Optional list of module prefixes that are allowed + for step class loading. If provided, step classes must come from modules + starting with one of these prefixes. + """ + inner_step_type = get_step_class_from_qualified_name( + self.inner_step_python_type, + allowed_module_prefixes=allowed_module_prefixes, + ) return KernelProcessStepInfo(inner_step_type=inner_step_type, state=self.state, output_edges=self.edges) @classmethod @@ -46,9 +53,3 @@ def from_kernel_step_info(cls, kernel_step_info: KernelProcessStepInfo) -> "Dapr state=kernel_step_info.state, edges={key: list(value) for key, value in kernel_step_info.edges.items()}, ) - - def _get_class_from_string(self, full_class_name: str): - """Gets a class from a string.""" - module_name, class_name = full_class_name.rsplit(".", 1) - module = importlib.import_module(module_name) - return getattr(module, class_name) diff --git a/python/semantic_kernel/processes/step_utils.py b/python/semantic_kernel/processes/step_utils.py index 85754c5f2435..66c72008ae75 100644 --- a/python/semantic_kernel/processes/step_utils.py +++ b/python/semantic_kernel/processes/step_utils.py @@ -1,9 +1,14 @@ # Copyright (c) Microsoft. All rights reserved. +import importlib +import inspect +from collections.abc import Sequence from typing import Any +from semantic_kernel.exceptions.process_exceptions import ProcessInvalidConfigurationException from semantic_kernel.functions.kernel_function import KernelFunction from semantic_kernel.processes.kernel_process.kernel_process_message_channel import KernelProcessMessageChannel +from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep from semantic_kernel.processes.kernel_process.kernel_process_step_context import KernelProcessStepContext from semantic_kernel.utils.feature_stage_decorator import experimental @@ -37,3 +42,80 @@ def find_input_channels( def get_fully_qualified_name(cls) -> str: """Gets the fully qualified name of a class.""" return f"{cls.__module__}.{cls.__name__}" + + +@experimental +def get_step_class_from_qualified_name( + full_class_name: str, + allowed_module_prefixes: Sequence[str] | None = None, +) -> type[KernelProcessStep]: + """Loads and validates a KernelProcessStep class from a fully qualified name. + + This function validates that the loaded class is a proper subclass of + KernelProcessStep, preventing instantiation of arbitrary classes. + + Args: + full_class_name: The fully qualified class name in Python import notation + (e.g., 'mypackage.mymodule.MyStep'). The module must be importable + from the current Python environment. + allowed_module_prefixes: Optional list of module prefixes that are allowed + to be imported. If provided, the module must start with one of these + prefixes. This check is performed BEFORE import to prevent execution + of module-level code in unauthorized modules. If None or empty, any + module is allowed. + + Returns: + The validated class type that is a subclass of KernelProcessStep + + Raises: + ProcessInvalidConfigurationException: Raised when: + - The class name format is invalid (missing module separator) + - The module is not in the allowed prefixes list (if provided) + - The module cannot be imported + - The class attribute doesn't exist in the module + - The attribute is not a class type + - The class is not a subclass of KernelProcessStep + """ + if not full_class_name or "." not in full_class_name: + raise ProcessInvalidConfigurationException( + f"Invalid step class name format: '{full_class_name}'. " + "Expected a fully qualified name like 'module.ClassName'." + ) + + module_name, class_name = full_class_name.rsplit(".", 1) + + if not module_name or not class_name: + raise ProcessInvalidConfigurationException( + f"Invalid step class name format: '{full_class_name}'. Module name and class name cannot be empty." + ) + + # Check module allowlist BEFORE import to prevent module-level code execution + if allowed_module_prefixes and not any(module_name.startswith(prefix) for prefix in allowed_module_prefixes): + raise ProcessInvalidConfigurationException( + f"Module '{module_name}' is not in the allowed module prefixes: {allowed_module_prefixes}. " + f"Step class '{full_class_name}' cannot be loaded." + ) + + try: + module = importlib.import_module(module_name) + except ImportError as e: + raise ProcessInvalidConfigurationException( + f"Unable to import module '{module_name}' for step class '{full_class_name}': {e}" + ) from e + + try: + cls = getattr(module, class_name) + except AttributeError as e: + raise ProcessInvalidConfigurationException( + f"Class '{class_name}' not found in module '{module_name}': {e}" + ) from e + + if not inspect.isclass(cls): + raise ProcessInvalidConfigurationException(f"'{full_class_name}' is not a class type, got {type(cls).__name__}") + + if not issubclass(cls, KernelProcessStep): + raise ProcessInvalidConfigurationException( + f"Step class '{full_class_name}' must be a subclass of KernelProcessStep. Got: {cls.__bases__}" + ) + + return cls diff --git a/python/tests/unit/processes/dapr_runtime/test_dapr_kernel_process_context.py b/python/tests/unit/processes/dapr_runtime/test_dapr_kernel_process_context.py index 487958430032..c344f117a817 100644 --- a/python/tests/unit/processes/dapr_runtime/test_dapr_kernel_process_context.py +++ b/python/tests/unit/processes/dapr_runtime/test_dapr_kernel_process_context.py @@ -11,11 +11,14 @@ from semantic_kernel.processes.kernel_process.kernel_process import KernelProcess from semantic_kernel.processes.kernel_process.kernel_process_event import KernelProcessEvent from semantic_kernel.processes.kernel_process.kernel_process_state import KernelProcessState +from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep from semantic_kernel.processes.kernel_process.kernel_process_step_info import KernelProcessStepInfo from semantic_kernel.processes.kernel_process.kernel_process_step_state import KernelProcessStepState -class DummyInnerStepType: +class DummyInnerStepType(KernelProcessStep): + """A valid KernelProcessStep subclass for testing.""" + pass diff --git a/python/tests/unit/processes/dapr_runtime/test_step_class_loading.py b/python/tests/unit/processes/dapr_runtime/test_step_class_loading.py new file mode 100644 index 000000000000..4fd43f957b79 --- /dev/null +++ b/python/tests/unit/processes/dapr_runtime/test_step_class_loading.py @@ -0,0 +1,176 @@ +# Copyright (c) Microsoft. All rights reserved. + +import pytest + +from semantic_kernel.exceptions.process_exceptions import ProcessInvalidConfigurationException +from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep +from semantic_kernel.processes.step_utils import get_step_class_from_qualified_name + + +class MockValidStep(KernelProcessStep): + """A valid KernelProcessStep subclass for testing.""" + + pass + + +class NotAStep: + """A class that is NOT a KernelProcessStep subclass.""" + + pass + + +def test_valid_step_class_loads_successfully(): + """Test that a valid KernelProcessStep subclass loads correctly.""" + full_name = f"{MockValidStep.__module__}.{MockValidStep.__name__}" + result = get_step_class_from_qualified_name(full_name) + assert result is MockValidStep + assert issubclass(result, KernelProcessStep) + + +def test_invalid_format_no_dots_raises_exception(): + """Test that a class name without dots raises ProcessInvalidConfigurationException.""" + with pytest.raises(ProcessInvalidConfigurationException, match="Invalid step class name format"): + get_step_class_from_qualified_name("NoDots") + + +def test_empty_module_name_raises_exception(): + """Test that empty module name (e.g., '.ClassName') raises ProcessInvalidConfigurationException.""" + with pytest.raises(ProcessInvalidConfigurationException, match="Module name and class name cannot be empty"): + get_step_class_from_qualified_name(".ClassName") + + +def test_empty_class_name_raises_exception(): + """Test that empty class name (e.g., 'module.') raises ProcessInvalidConfigurationException.""" + with pytest.raises(ProcessInvalidConfigurationException, match="Module name and class name cannot be empty"): + get_step_class_from_qualified_name("module.") + + +def test_empty_string_raises_exception(): + """Test that an empty string raises ProcessInvalidConfigurationException.""" + with pytest.raises(ProcessInvalidConfigurationException, match="Invalid step class name format"): + get_step_class_from_qualified_name("") + + +def test_none_like_empty_raises_exception(): + """Test that None-like values raise ProcessInvalidConfigurationException.""" + with pytest.raises(ProcessInvalidConfigurationException, match="Invalid step class name format"): + get_step_class_from_qualified_name(None) # type: ignore + + +def test_nonexistent_module_raises_exception(): + """Test that a non-existent module raises ProcessInvalidConfigurationException.""" + with pytest.raises(ProcessInvalidConfigurationException, match="Unable to import module"): + get_step_class_from_qualified_name("nonexistent_module_xyz123.SomeClass") + + +def test_nonexistent_class_in_valid_module_raises_exception(): + """Test that a non-existent class in a valid module raises ProcessInvalidConfigurationException.""" + with pytest.raises(ProcessInvalidConfigurationException, match="not found in module"): + get_step_class_from_qualified_name("semantic_kernel.kernel.NonExistentClass12345") + + +def test_non_class_attribute_raises_exception(): + """Test that loading a non-class attribute (function) raises ProcessInvalidConfigurationException.""" + with pytest.raises(ProcessInvalidConfigurationException, match="is not a class type"): + get_step_class_from_qualified_name("semantic_kernel.processes.step_utils.get_fully_qualified_name") + + +def test_non_step_class_raises_exception(): + """Test that a class not inheriting from KernelProcessStep raises exception.""" + full_name = f"{NotAStep.__module__}.{NotAStep.__name__}" + with pytest.raises(ProcessInvalidConfigurationException, match="must be a subclass of KernelProcessStep"): + get_step_class_from_qualified_name(full_name) + + +def test_builtin_class_raises_exception(): + """Test that built-in classes like str raise exception.""" + with pytest.raises(ProcessInvalidConfigurationException, match="must be a subclass of KernelProcessStep"): + get_step_class_from_qualified_name("builtins.str") + + +def test_os_system_prevented(): + """Test that os.system (a dangerous function, not a class) is prevented.""" + with pytest.raises(ProcessInvalidConfigurationException, match="is not a class type"): + get_step_class_from_qualified_name("os.system") + + +def test_arbitrary_class_prevented(): + """Test that arbitrary classes like subprocess.Popen are prevented.""" + with pytest.raises(ProcessInvalidConfigurationException, match="must be a subclass of KernelProcessStep"): + get_step_class_from_qualified_name("subprocess.Popen") + + +def test_kernel_class_prevented(): + """Test that even internal SK classes that aren't steps are prevented.""" + with pytest.raises(ProcessInvalidConfigurationException, match="must be a subclass of KernelProcessStep"): + get_step_class_from_qualified_name("semantic_kernel.kernel.Kernel") + + +# Module allowlist tests + + +def test_allowlist_permits_valid_module(): + """Test that a valid step class from an allowed module loads successfully.""" + full_name = f"{MockValidStep.__module__}.{MockValidStep.__name__}" + # Allow the test module (use actual module name from __module__) + module_prefix = MockValidStep.__module__ + result = get_step_class_from_qualified_name( + full_name, + allowed_module_prefixes=[module_prefix], + ) + assert result is MockValidStep + + +def test_allowlist_blocks_disallowed_module(): + """Test that a module not in the allowlist is blocked BEFORE import.""" + with pytest.raises(ProcessInvalidConfigurationException, match="is not in the allowed module prefixes"): + get_step_class_from_qualified_name( + "os.path", + allowed_module_prefixes=["semantic_kernel.", "myapp."], + ) + + +def test_allowlist_blocks_dangerous_module(): + """Test that dangerous modules are blocked when allowlist is provided.""" + with pytest.raises(ProcessInvalidConfigurationException, match="is not in the allowed module prefixes"): + get_step_class_from_qualified_name( + "subprocess.Popen", + allowed_module_prefixes=["semantic_kernel."], + ) + + +def test_empty_allowlist_allows_all(): + """Test that an empty allowlist allows any module (current behavior).""" + full_name = f"{MockValidStep.__module__}.{MockValidStep.__name__}" + result = get_step_class_from_qualified_name(full_name, allowed_module_prefixes=[]) + assert result is MockValidStep + + +def test_none_allowlist_allows_all(): + """Test that None allowlist allows any module (default behavior).""" + full_name = f"{MockValidStep.__module__}.{MockValidStep.__name__}" + result = get_step_class_from_qualified_name(full_name, allowed_module_prefixes=None) + assert result is MockValidStep + + +def test_allowlist_prefix_matching(): + """Test that allowlist uses prefix matching correctly.""" + full_name = f"{MockValidStep.__module__}.{MockValidStep.__name__}" + # Use a prefix of the actual module name + module_prefix = MockValidStep.__module__[:4] # First 4 chars as prefix + result = get_step_class_from_qualified_name( + full_name, + allowed_module_prefixes=[module_prefix], + ) + assert result is MockValidStep + + +def test_allowlist_multiple_prefixes(): + """Test that multiple allowed prefixes work correctly.""" + full_name = f"{MockValidStep.__module__}.{MockValidStep.__name__}" + module_prefix = MockValidStep.__module__ + result = get_step_class_from_qualified_name( + full_name, + allowed_module_prefixes=["semantic_kernel.", "myapp.", module_prefix], + ) + assert result is MockValidStep