diff --git a/src/sap_cloud_sdk/dms/client.py b/src/sap_cloud_sdk/dms/client.py index 5bcaa785..5b6dc3eb 100644 --- a/src/sap_cloud_sdk/dms/client.py +++ b/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, diff --git a/tests/dms/unit/test_client_admin.py b/tests/dms/unit/test_client_admin.py index 5a74b576..83398a7e 100644 --- a/tests/dms/unit/test_client_admin.py +++ b/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) diff --git a/tests/dms/unit/test_client_cmis.py b/tests/dms/unit/test_client_cmis.py index d93687d1..b52d454b 100644 --- a/tests/dms/unit/test_client_cmis.py +++ b/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")