fix(skills): validate skill_id path segments before GCS blob lookup

src/google/adk/skills/_utils.py

@@ -33,6 +33,31 @@
 # Bounds on a skill archive, which may come from a remote registry and is
 # untrusted until it has been loaded. They are generous relative to any
 # realistic skill; the toolset already warns about payloads over 16 MB.
+def _validate_path_segment(value: str, field_name: str) -> None:
+  """Rejects values that could alter a storage blob path.
+
+  Args:
+    value: The caller-supplied identifier.
+    field_name: Human-readable field name used in error messages.
+
+  Raises:
+    ValueError: If the value contains path separators, traversal segments,
+      or null bytes.
+  """
+  if not value:
+    raise ValueError(f"{field_name} must not be empty.")
+  if "\x00" in value:
+    raise ValueError(f"{field_name} must not contain null bytes.")
+  if "\\" in value:
+    raise ValueError(
+        f"{field_name} {value!r} must not contain path separators."
+    )
+  if value in (".", ".."):
+    raise ValueError(
+        f"{field_name} {value!r} must not contain traversal segments."
+    )
+
+
 _MAX_ZIP_ENTRIES = 2000
 _MAX_ZIP_UNCOMPRESSED_BYTES = 32 * 1024 * 1024
 # How much of a member is decompressed per step. Reading in steps keeps the
@@ -623,6 +648,15 @@ def _load_skill_from_gcs_dir(
   client = storage.Client(project=project_id, credentials=credentials)
   bucket = client.bucket(bucket_name)
 
+  # skill_id identifies which of potentially many skill directories under
+  # a shared bucket gets loaded. An application may resolve it from a
+  # caller- or model-selected skill name rather than a fixed,
+  # developer-authored constant, so each segment is validated the same
+  # way app_name/eval_set_id are validated in the evaluation GCS managers
+  # before being interpolated into a blob prefix.
+  for segment in skill_id.strip("/").split("/"):
+    _validate_path_segment(segment, "skill_id")
+
   base_prefix = skills_base_path.strip("/")
   if base_prefix:
     base_prefix += "/"

tests/unittests/skills/test__utils.py

@@ -421,6 +421,31 @@ def list_blobs_side_effect(prefix=None):
   assert skill.resources.get_reference("ref1.md") == "ref1 content"
 
 
+@mock.patch("google.cloud.storage.Client")
+@pytest.mark.parametrize(
+    "skill_id",
+    [
+        "../../other-tenant/secrets",
+        "skills/../../other-tenant/secrets",
+        "..",
+        "skills/..",
+    ],
+)
+def test__load_skill_from_gcs_dir_rejects_traversal(
+    mock_client_class, skill_id
+):
+  """A traversal-shaped skill_id must be rejected before any blob lookup."""
+  mock_client = mock.MagicMock()
+  mock_client_class.return_value = mock_client
+  mock_bucket = mock.MagicMock()
+  mock_client.bucket.return_value = mock_bucket
+
+  with pytest.raises(ValueError, match="skill_id"):
+    _load_skill_from_gcs_dir("my-bucket", skill_id, skills_base_path="skills")
+
+  mock_bucket.blob.assert_not_called()
+
+
 @mock.patch("google.cloud.storage.Client")
 def test__load_skill_from_gcs_dir_binary_resources(mock_client_class):
   """Tests that non-UTF-8 GCS blobs are loaded as bytes, and scripts skipped."""