Validate FileArtifactService scope identifiers before building storage paths

src/google/adk/artifacts/file_artifact_service.py

@@ -85,6 +85,22 @@ def _to_posix_path(path_value: str) -> PurePosixPath:
   return PurePosixPath(path_value)
 
 
+def _validate_scope_identifier(identifier_name: str, identifier: str) -> str:
+  """Validates user/session identifiers before using them in paths."""
+  pure_path = _to_posix_path(identifier)
+  if pure_path.is_absolute() or len(pure_path.parts) != 1:
+    raise InputValidationError(
+        f"{identifier_name} {identifier!r} must be a single path component."
+    )
+
+  component = pure_path.parts[0]
+  if component in {"", ".", ".."}:
+    raise InputValidationError(
+        f"{identifier_name} {identifier!r} is not a valid storage scope."
+    )
+  return component
+
+
 def _resolve_scoped_artifact_path(
     scope_root: Path, filename: str
 ) -> tuple[Path, Path]:
@@ -145,7 +161,12 @@ def _user_artifacts_dir(base_root: Path) -> Path:
 
 def _session_artifacts_dir(base_root: Path, session_id: str) -> Path:
   """Returns the path that stores session-scoped artifacts."""
-  return base_root / "sessions" / session_id / "artifacts"
+  return (
+      base_root
+      / "sessions"
+      / _validate_scope_identifier("session_id", session_id)
+      / "artifacts"
+  )
 
 
 def _versions_dir(artifact_dir: Path) -> Path:
@@ -220,7 +241,11 @@ def __init__(self, root_dir: Path | str):
 
   def _base_root(self, user_id: str, /) -> Path:
     """Returns the artifacts root directory for a user."""
-    return self.root_dir / "users" / user_id
+    return (
+        self.root_dir
+        / "users"
+        / _validate_scope_identifier("user_id", user_id)
+    )
 
   def _scope_root(
       self,

tests/unittests/artifacts/test_artifact_service.py

@@ -769,6 +769,66 @@ async def test_file_save_artifact_rejects_absolute_path_within_scope(tmp_path):
     )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "user_id",
+    [
+        "../escape-user",
+        "..\\escape-user",
+        "nested/user",
+        "nested\\user",
+        "/absolute-user",
+        "C:\\absolute-user",
+        ".",
+        "..",
+    ],
+)
+async def test_file_save_artifact_rejects_invalid_user_scope_identifiers(
+    tmp_path, user_id
+):
+  """FileArtifactService rejects user IDs that alter the scope path."""
+  artifact_service = FileArtifactService(root_dir=tmp_path / "artifacts")
+  with pytest.raises(InputValidationError):
+    await artifact_service.save_artifact(
+        app_name="myapp",
+        user_id=user_id,
+        session_id="sess123",
+        filename="proof.txt",
+        artifact=types.Part(text="content"),
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "session_id",
+    [
+        "../escape-session",
+        "..\\escape-session",
+        "../../../escape-session",
+        "..\\..\\..\\escape-session",
+        "nested/session",
+        "nested\\session",
+        "/absolute-session",
+        "C:\\absolute-session",
+        ".",
+        "..",
+    ],
+)
+async def test_file_save_artifact_rejects_invalid_session_scope_identifiers(
+    tmp_path, session_id
+):
+  """FileArtifactService rejects session IDs that alter the scope path."""
+  artifact_service = FileArtifactService(root_dir=tmp_path / "artifacts")
+  with pytest.raises(InputValidationError):
+    await artifact_service.save_artifact(
+        app_name="myapp",
+        user_id="user123",
+        session_id=session_id,
+        filename="proof.txt",
+        artifact=types.Part(text="content"),
+    )
+
+
 class TestEnsurePart:
   """Tests for the ensure_part normalization helper."""