session manager - prevent file path injection

Author: pgrayy

src/strands/agent/agent.py

@@ -40,6 +40,7 @@
 from ..types.exceptions import ContextWindowOverflowException
 from ..types.tools import ToolResult, ToolUse
 from ..types.traces import AttributeValue
+from . import identifier
 from .agent_result import AgentResult
 from .conversation_manager import (
     ConversationManager,
@@ -249,12 +250,15 @@ def __init__(
                 Defaults to None.
             session_manager: Manager for handling agent sessions including conversation history and state.
                 If provided, enables session-based persistence and state management.
+
+        Raises:
+            ValueError: If agent id contains path separators.
         """
         self.model = BedrockModel() if not model else BedrockModel(model_id=model) if isinstance(model, str) else model
         self.messages = messages if messages is not None else []
 
         self.system_prompt = system_prompt
-        self.agent_id = agent_id or _DEFAULT_AGENT_ID
+        self.agent_id = identifier.validate(agent_id or _DEFAULT_AGENT_ID)
         self.name = name or _DEFAULT_AGENT_NAME
         self.description = description
 

src/strands/agent/identifier.py

@@ -0,0 +1,21 @@
+"""Agent identifier utilities."""
+
+import os
+
+
+def validate(agent_id: str) -> str:
+    """Validate agent id.
+
+    Args:
+        agent_id: Id to validate.
+
+    Returns:
+        Validated id.
+
+    Raises:
+        ValueError: If id contains path separators.
+    """
+    if os.path.basename(agent_id) != agent_id:
+        raise ValueError(f"agent_id={agent_id} | agent id cannot contain path separators")
+
+    return agent_id

src/strands/session/file_session_manager.py

@@ -7,8 +7,10 @@
 import tempfile
 from typing import Any, Optional, cast
 
+from ..agent.identifier import validate as validate_agent_id
 from ..types.exceptions import SessionException
 from ..types.session import Session, SessionAgent, SessionMessage
+from .identifier import validate as validate_session_id
 from .repository_session_manager import RepositorySessionManager
 from .session_repository import SessionRepository
 
@@ -40,8 +42,9 @@ def __init__(self, session_id: str, storage_dir: Optional[str] = None, **kwargs:
         """Initialize FileSession with filesystem storage.
 
         Args:
-            session_id: ID for the session
-            storage_dir: Directory for local filesystem storage (defaults to temp dir)
+            session_id: ID for the session.
+                ID is not allowed to contain path separators (e.g., a/b).
+            storage_dir: Directory for local filesystem storage (defaults to temp dir).
             **kwargs: Additional keyword arguments for future extensibility.
         """
         self.storage_dir = storage_dir or os.path.join(tempfile.gettempdir(), "strands/sessions")
@@ -50,12 +53,29 @@ def __init__(self, session_id: str, storage_dir: Optional[str] = None, **kwargs:
         super().__init__(session_id=session_id, session_repository=self)
 
     def _get_session_path(self, session_id: str) -> str:
-        """Get session directory path."""
+        """Get session directory path.
+
+        Args:
+            session_id: ID for the session.
+
+        Raises:
+            ValueError: If session id contains a path separator.
+        """
+        session_id = validate_session_id(session_id)
         return os.path.join(self.storage_dir, f"{SESSION_PREFIX}{session_id}")
 
     def _get_agent_path(self, session_id: str, agent_id: str) -> str:
-        """Get agent directory path."""
+        """Get agent directory path.
+
+        Args:
+            session_id: ID for the session.
+            agent_id: ID for the agent.
+
+        Raises:
+            ValueError: If session id or agent id contains a path separator.
+        """
         session_path = self._get_session_path(session_id)
+        agent_id = validate_agent_id(agent_id)
         return os.path.join(session_path, "agents", f"{AGENT_PREFIX}{agent_id}")
 
     def _get_message_path(self, session_id: str, agent_id: str, message_id: int) -> str:

src/strands/session/identifier.py

@@ -0,0 +1,21 @@
+"""Session identifier utilities."""
+
+import os
+
+
+def validate(session_id: str) -> str:
+    """Validate session id.
+
+    Args:
+        session_id: Id to validate.
+
+    Returns:
+        Validated id.
+
+    Raises:
+        ValueError: If id contains path separators.
+    """
+    if os.path.basename(session_id) != session_id:
+        raise ValueError(f"session_id={session_id} | session id cannot contain path separators")
+
+    return session_id

src/strands/session/s3_session_manager.py

@@ -8,8 +8,10 @@
 from botocore.config import Config as BotocoreConfig
 from botocore.exceptions import ClientError
 
+from ..agent.identifier import validate as validate_agent_id
 from ..types.exceptions import SessionException
 from ..types.session import Session, SessionAgent, SessionMessage
+from .identifier import validate as validate_session_id
 from .repository_session_manager import RepositorySessionManager
 from .session_repository import SessionRepository
 
@@ -51,6 +53,7 @@ def __init__(
 
         Args:
             session_id: ID for the session
+                ID is not allowed to contain path separators (e.g., a/b).
             bucket: S3 bucket name (required)
             prefix: S3 key prefix for storage organization
             boto_session: Optional boto3 session
@@ -79,12 +82,29 @@ def __init__(
         super().__init__(session_id=session_id, session_repository=self)
 
     def _get_session_path(self, session_id: str) -> str:
-        """Get session S3 prefix."""
+        """Get session S3 prefix.
+
+        Args:
+            session_id: ID for the session.
+
+        Raises:
+            ValueError: If session id contains a path separator.
+        """
+        session_id = validate_session_id(session_id)
         return f"{self.prefix}/{SESSION_PREFIX}{session_id}/"
 
     def _get_agent_path(self, session_id: str, agent_id: str) -> str:
-        """Get agent S3 prefix."""
+        """Get agent S3 prefix.
+
+        Args:
+            session_id: ID for the session.
+            agent_id: ID for the agent.
+
+        Raises:
+            ValueError: If session id or agent id contains a path separator.
+        """
         session_path = self._get_session_path(session_id)
+        agent_id = validate_agent_id(agent_id)
         return f"{session_path}agents/{AGENT_PREFIX}{agent_id}/"
 
     def _get_message_path(self, session_id: str, agent_id: str, message_id: int) -> str:

tests/strands/agent/test_agent.py

@@ -250,6 +250,18 @@ def test_agent__init__deeply_nested_tools(tool_decorated, tool_module, tool_impo
     assert tru_tool_names == exp_tool_names
 
 
+@pytest.mark.parametrize(
+    "agent_id",
+    [
+        "a/../b",
+        "a/b",
+    ],
+)
+def test_agent__init__invalid_id(agent_id):
+    with pytest.raises(ValueError):
+        Agent(agent_id=agent_id)
+
+
 def test_agent__call__(
     mock_model,
     system_prompt,

tests/strands/agent/test_identifier.py

@@ -0,0 +1,21 @@
+import pytest
+
+from strands.agent import identifier
+
+
+def test_validate():
+    tru_id = identifier.validate("abc")
+    exp_id = "abc"
+    assert tru_id == exp_id
+
+
+@pytest.mark.parametrize(
+    "agent_id",
+    [
+        "a/../b",
+        "a/b",
+    ],
+)
+def test_validate_invalid(agent_id):
+    with pytest.raises(ValueError):
+        identifier.validate(agent_id)