fix(dms): prevent URL path injection in DMS client

src/sap_cloud_sdk/dms/client.py

@@ -1,5 +1,7 @@
 import logging
+import re
 from typing import BinaryIO, Dict, List, Optional, Union
+from urllib.parse import quote
 from requests import Response
 from sap_cloud_sdk.dms.model import (
     DMSCredentials,
@@ -33,6 +35,31 @@
 # ---------------------------------------------------------------------------
 _REPOSITORIES = "/rest/v2/repositories"
 _CONFIGS = "/rest/v2/configs"
+
+_SAFE_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$")
+
+
+def _validate_id(value: str, param_name: str) -> None:
+    """Raise ValueError if *value* is empty or contains URL path metacharacters."""
+    if not value or not value.strip():
+        raise ValueError(f"{param_name} must not be empty")
+    if not _SAFE_ID_RE.match(value):
+        raise ValueError(
+            f"{param_name} contains characters not allowed in a URL path segment: {value!r}"
+        )
+
+
+def _safe_cmis_path(path: str) -> str:
+    """Percent-encode a CMIS browse path; raise ValueError on dot-segment traversal."""
+    segments = [s for s in path.split("/") if s]
+    for seg in segments:
+        if seg in (".", ".."):
+            raise ValueError(
+                f"Path traversal segment {seg!r} is not allowed in a CMIS path"
+            )
+    return "/".join(quote(seg, safe="") for seg in segments)
+
+
 # ---------------------------------------------------------------------------
 
 
@@ -207,6 +234,7 @@ def get_repository(
             DMSRuntimeException: If the server encounters an internal error.
         """
         logger.info("Fetching repository '%s'", repo_id)
+        _validate_id(repo_id, "repo_id")
         response = self._http.get(
             path=f"{_REPOSITORIES}/{repo_id}",
             tenant_subdomain=tenant,
@@ -239,8 +267,7 @@ def update_repository(
             DMSPermissionDeniedException: If the access token is invalid or expired.
             DMSRuntimeException: If the server encounters an internal error.
         """
-        if not repo_id or not repo_id.strip():
-            raise ValueError("repo_id must not be empty")
+        _validate_id(repo_id, "repo_id")
         logger.info("Updating repository '%s'", repo_id)
         response = self._http.put(
             path=f"{_REPOSITORIES}/{repo_id}",
@@ -272,6 +299,7 @@ def delete_repository(
             DMSPermissionDeniedException: If the access token is invalid or expired.
             DMSRuntimeException: If the server encounters an internal error.
         """
+        _validate_id(repo_id, "repo_id")
         self._http.delete(
             path=f"{_REPOSITORIES}/{repo_id}",
             tenant_subdomain=tenant,
@@ -363,8 +391,7 @@ def update_config(
             DMSPermissionDeniedException: If the access token is invalid or expired.
             DMSRuntimeException: If the server encounters an internal error.
         """
-        if not config_id or not config_id.strip():
-            raise ValueError("config_id must not be empty")
+        _validate_id(config_id, "config_id")
         logger.info("Updating config '%s'", config_id)
         response = self._http.put(
             path=f"{_CONFIGS}/{config_id}",
@@ -396,8 +423,7 @@ def delete_config(
             DMSPermissionDeniedException: If the access token is invalid or expired.
             DMSRuntimeException: If the server encounters an internal error.
         """
-        if not config_id or not config_id.strip():
-            raise ValueError("config_id must not be empty")
+        _validate_id(config_id, "config_id")
         logger.info("Deleting config '%s'", config_id)
         self._http.delete(
             path=f"{_CONFIGS}/{config_id}",
@@ -411,9 +437,10 @@ def delete_config(
 
     @staticmethod
     def _browser_url(repository_id: str, path: Optional[str] = None) -> str:
+        _validate_id(repository_id, "repository_id")
         base = f"/browser/{repository_id}/root"
         if path:
-            return f"{base}/{path.lstrip('/')}"
+            return f"{base}/{_safe_cmis_path(path)}"
         return base
 
     # ==================================================================
@@ -1083,6 +1110,8 @@ def restore_object(
             DMSPermissionDeniedException: If the access token is invalid.
             DMSRuntimeException: If the server encounters an internal error.
         """
+        _validate_id(repository_id, "repository_id")
+        _validate_id(object_id, "object_id")
         path = f"{_REPOSITORIES}/{repository_id}/deleted/objects/{object_id}/restore"
         logger.info("Restoring object '%s' in repo '%s'", object_id, repository_id)
         response = self._http.post(
@@ -1208,6 +1237,7 @@ def cmis_query(
         params.update(opts.to_query_params())
 
         logger.info("Executing CMIS query in repo '%s'", repository_id)
+        _validate_id(repository_id, "repository_id")
         response = self._http.get(
             f"/browser/{repository_id}",
             params=params,

tests/dms/unit/test_client_admin.py

@@ -443,3 +443,80 @@ def test_with_tenant_and_user_claim(self, client):
         call_args = client._mock_http.delete.call_args
         assert call_args[1]["tenant_subdomain"] == "t1"
         assert call_args[1]["user_claim"] is claim
+
+
+# ---------------------------------------------------------------
+# URL path injection — security regression tests
+# ---------------------------------------------------------------
+
+
+class TestUrlInjectionPrevention:
+    """Regression tests: caller-supplied IDs with path metacharacters must be rejected.
+
+    Slashes, dot-segments, and query characters in an ID could redirect the
+    authenticated request to unintended endpoints before it reaches the HTTP layer.
+    """
+
+    # repo_id sinks
+    @pytest.mark.parametrize(
+        "bad_id",
+        [
+            "../../../rest/v2/repositories/other",
+            "repo/../admin",
+            "repo?injected=yes",
+            "repo#fragment",
+            "/etc/passwd",
+        ],
+    )
+    def test_get_repository_rejects_injection(self, client, bad_id):
+        with pytest.raises(ValueError, match="contains characters not allowed"):
+            client.get_repository(bad_id)
+
+    @pytest.mark.parametrize(
+        "bad_id",
+        [
+            "../../../rest/v2/repositories/other",
+            "repo/../admin",
+            "repo?injected=yes",
+        ],
+    )
+    def test_update_repository_rejects_injection(self, client, bad_id):
+        request = UpdateRepoRequest(description="x")
+        with pytest.raises(ValueError, match="contains characters not allowed"):
+            client.update_repository(bad_id, request)
+
+    @pytest.mark.parametrize(
+        "bad_id",
+        [
+            "../../../rest/v2/repositories/other",
+            "repo/../admin",
+        ],
+    )
+    def test_delete_repository_rejects_injection(self, client, bad_id):
+        with pytest.raises(ValueError, match="contains characters not allowed"):
+            client.delete_repository(bad_id)
+
+    # config_id sinks
+    @pytest.mark.parametrize(
+        "bad_id",
+        [
+            "../../../rest/v2/configs/other",
+            "cfg/../admin",
+            "cfg?injected=yes",
+        ],
+    )
+    def test_update_config_rejects_injection(self, client, bad_id):
+        request = UpdateConfigRequest(id="x", config_name="n", config_value="v")
+        with pytest.raises(ValueError, match="contains characters not allowed"):
+            client.update_config(bad_id, request)
+
+    @pytest.mark.parametrize(
+        "bad_id",
+        [
+            "../../../rest/v2/configs/other",
+            "cfg/../admin",
+        ],
+    )
+    def test_delete_config_rejects_injection(self, client, bad_id):
+        with pytest.raises(ValueError, match="contains characters not allowed"):
+            client.delete_config(bad_id)

tests/dms/unit/test_client_cmis.py

@@ -1277,3 +1277,62 @@ def test_with_user_claim(self, client):
         )
 
         assert client._mock_http.get.call_args[1]["user_claim"] is claim
+
+
+# ---------------------------------------------------------------
+# URL path injection — security regression tests
+# ---------------------------------------------------------------
+
+
+class TestUrlInjectionPrevention:
+    """Regression tests: URL path injection must be rejected before hitting HTTP."""
+
+    _TRAVERSAL_IDS = [
+        "../../../browser/evil-repo/root",
+        "repo/../../../admin",
+        "repo?injected=yes",
+        "repo#fragment",
+        "/etc/passwd",
+    ]
+
+    # _browser_url: repository_id injection
+    @pytest.mark.parametrize("bad_id", _TRAVERSAL_IDS)
+    def test_browser_url_rejects_repository_id_injection(self, bad_id):
+        with pytest.raises(ValueError, match="contains characters not allowed"):
+            DMSClient._browser_url(bad_id)
+
+    # _browser_url: path dot-segment traversal
+    @pytest.mark.parametrize(
+        "bad_path",
+        [
+            "../../../rest/v2/repositories",
+            "docs/../../../admin",
+            "docs/./secret",
+        ],
+    )
+    def test_browser_url_rejects_dotdot_in_path(self, bad_path):
+        with pytest.raises(ValueError, match="traversal segment"):
+            DMSClient._browser_url("repo1", bad_path)
+
+    # _browser_url: query/fragment in path segment
+    def test_browser_url_encodes_special_chars_in_path_segment(self):
+        url = DMSClient._browser_url("repo1", "docs/file name.pdf")
+        assert "file%20name.pdf" in url
+
+    # restore_object: repository_id injection
+    @pytest.mark.parametrize("bad_id", _TRAVERSAL_IDS)
+    def test_restore_object_rejects_repository_id_injection(self, client, bad_id):
+        with pytest.raises(ValueError, match="contains characters not allowed"):
+            client.restore_object(bad_id, "valid-object-id")
+
+    # restore_object: object_id injection
+    @pytest.mark.parametrize("bad_id", _TRAVERSAL_IDS)
+    def test_restore_object_rejects_object_id_injection(self, client, bad_id):
+        with pytest.raises(ValueError, match="contains characters not allowed"):
+            client.restore_object("valid-repo-id", bad_id)
+
+    # cmis_query: repository_id injection
+    @pytest.mark.parametrize("bad_id", _TRAVERSAL_IDS)
+    def test_cmis_query_rejects_repository_id_injection(self, client, bad_id):
+        with pytest.raises(ValueError, match="contains characters not allowed"):
+            client.cmis_query(bad_id, "SELECT * FROM cmis:document")