fix(destination): validate and encode resource names in URL path segments

CHANGELOG.md

@@ -0,0 +1,20 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
+
+## [Unreleased]
+
+### Security
+
+- Destination, fragment, and certificate resource names are now validated against an
+  allowlist grammar at every public SDK method boundary. Names must match
+  `[A-Za-z0-9][A-Za-z0-9._\-]{0,199}`. A `ValueError` is raised for invalid names
+  before any OAuth token is fetched or HTTP request is sent. This prevents path
+  traversal via crafted resource names in URL path segments.
+- `get_destination()` auto-parses a `@ConsumptionLevel` suffix in the name string for
+  backward compatibility (e.g. `"my-dest@provider_subaccount"` continues to work without
+  changes; the `level=` parameter is the canonical form). No breaking changes.
+- Added a defense-in-depth guard in `_request()` that rejects paths containing `..`
+  segments before issuing any HTTP request.

src/sap_cloud_sdk/destination/_http.py

@@ -30,6 +30,13 @@ def _request(
         all_headers.update(headers)
 
     normalized_path = f"/{path.lstrip('/')}"
+
+    # Defense-in-depth: reject any path that contains a '..' segment so that
+    # a crafted path cannot escape the Destination Service's own URL space,
+    # even if the client-layer validation is somehow bypassed.
+    path_segments = normalized_path.split("/")
+    if ".." in path_segments:
+        raise HttpError("Forbidden: path traversal detected")
     try:
         resp = http.request(
             method,

src/sap_cloud_sdk/destination/certificate_client.py

@@ -23,6 +23,10 @@
     PagedResult,
     parse_pagination_headers,
 )
+from sap_cloud_sdk.destination.utils._validation import (
+    encode_path_segment,
+    validate_resource_name,
+)
 
 _SUBACCOUNT_COLLECTION = "subaccountCertificates"
 _INSTANCE_COLLECTION = "instanceCertificates"
@@ -169,6 +173,7 @@ def get_instance_certificate(self, name: str) -> Optional[Certificate]:
         Raises:
             DestinationOperationError: If an HTTP error occurs or response parsing fails.
         """
+        validate_resource_name(name)
         try:
             return self._get_certificate(name=name, level=Level.SERVICE_INSTANCE)
         except HttpError as e:
@@ -203,6 +208,7 @@ def get_subaccount_certificate(
             DestinationOperationError: If tenant is missing for subscriber access strategies,
                                        on HTTP errors, or response parsing failures.
         """
+        validate_resource_name(name)
         try:
             return self._apply_access_strategy(
                 access_strategy=access_strategy,
@@ -239,6 +245,7 @@ def create_certificate(
             DestinationOperationError: For unexpected errors.
         """
         coll = self._sub_path_for_level(level)
+        validate_resource_name(certificate.name)
         body = certificate.to_dict()
 
         try:
@@ -280,6 +287,7 @@ def update_certificate(
             DestinationOperationError: For unexpected errors.
         """
         coll = self._sub_path_for_level(level)
+        validate_resource_name(certificate.name)
         body = certificate.to_dict()
 
         try:
@@ -317,12 +325,13 @@ def delete_certificate(
             DestinationOperationError: For unexpected errors.
         """
         coll = self._sub_path_for_level(level)
+        validate_resource_name(name)
 
         try:
             _request(
                 self._http,
                 HttpMethod.DELETE,
-                f"{API_V1}/{coll}/{name}",
+                f"{API_V1}/{coll}/{encode_path_segment(name)}",
                 tenant_subdomain=tenant,
             )
         except HttpError:
@@ -354,12 +363,13 @@ def get_certificate_labels(
         Raises:
             DestinationOperationError: If an HTTP error occurs or response parsing fails.
         """
+        validate_resource_name(name)
         try:
             path = self._sub_path_for_level(level)
             resp = _request(
                 self._http,
                 HttpMethod.GET,
-                f"{API_V1}/{path}/{name}/labels",
+                f"{API_V1}/{path}/{encode_path_segment(name)}/labels",
                 tenant_subdomain=tenant,
             )
             data = resp.json()
@@ -398,12 +408,13 @@ def update_certificate_labels(
             DestinationOperationError: For unexpected errors.
         """
         resolved_level = level or Level.SUB_ACCOUNT
+        validate_resource_name(name)
         try:
             path = self._sub_path_for_level(resolved_level)
             _request(
                 self._http,
                 HttpMethod.PUT,
-                f"{API_V1}/{path}/{name}/labels",
+                f"{API_V1}/{path}/{encode_path_segment(name)}/labels",
                 json=[lbl.to_dict() for lbl in labels],
                 tenant_subdomain=tenant,
             )
@@ -435,12 +446,13 @@ def patch_certificate_labels(
             DestinationOperationError: For unexpected errors.
         """
         resolved_level = level or Level.SUB_ACCOUNT
+        validate_resource_name(name)
         try:
             path = self._sub_path_for_level(resolved_level)
             _request(
                 self._http,
                 HttpMethod.PATCH,
-                f"{API_V1}/{path}/{name}/labels",
+                f"{API_V1}/{path}/{encode_path_segment(name)}/labels",
                 json=patch.to_dict(),
                 tenant_subdomain=tenant,
             )
@@ -478,7 +490,7 @@ def _get_certificate(
             resp = _request(
                 self._http,
                 HttpMethod.GET,
-                f"{API_V1}/{path}/{name}",
+                f"{API_V1}/{path}/{encode_path_segment(name)}",
                 tenant_subdomain=tenant_subdomain,
             )
             data = resp.json()

src/sap_cloud_sdk/destination/client.py

@@ -32,6 +32,10 @@
     PagedResult,
     parse_pagination_headers,
 )
+from sap_cloud_sdk.destination.utils._validation import (
+    encode_path_segment,
+    validate_resource_name,
+)
 
 T = TypeVar("T")
 
@@ -41,6 +45,59 @@
 _INSTANCE_COLLECTION = "instanceDestinations"
 
 
+def _split_name_and_level(
+    name: str,
+    level: Optional[ConsumptionLevel],
+) -> tuple[str, Optional[ConsumptionLevel]]:
+    """Auto-parse a legacy ``@ConsumptionLevel`` suffix from a destination name.
+
+    Allows callers that use the old pattern ``"my-dest@provider_subaccount"``
+    to keep working without changes.  The ``@``-suffix is extracted, validated
+    against the :class:`ConsumptionLevel` enum, and merged with the explicit
+    ``level`` parameter.
+
+    * If *name* contains no ``@``, returns ``(name, level)`` unchanged.
+    * If the suffix is not a recognised :class:`ConsumptionLevel` value,
+      raises :exc:`ValueError` with a clear message listing valid values.
+    * If both the suffix and the explicit *level* parameter are set to
+      **different** values, raises :exc:`ValueError` (conflicting levels).
+    * If both are set to the **same** value, the redundancy is accepted.
+
+    Args:
+        name:  Raw destination name, possibly containing an ``@level`` suffix.
+        level: Explicit level parameter passed by the caller (may be ``None``).
+
+    Returns:
+        ``(base_name, resolved_level)`` — the base name (before any ``@``) and
+        the resolved :class:`ConsumptionLevel` (or ``None`` if neither source
+        provided one).
+
+    Raises:
+        ValueError: If the suffix is not a valid :class:`ConsumptionLevel`, or
+            if the suffix and the explicit *level* parameter conflict.
+    """
+    if "@" not in name:
+        return name, level
+
+    base_name, _, level_hint = name.partition("@")
+    try:
+        parsed_level = ConsumptionLevel(level_hint)
+    except ValueError:
+        valid = [e.value for e in ConsumptionLevel]
+        raise ValueError(
+            f"Invalid resource name {name!r}: '@' is only allowed as a "
+            f"ConsumptionLevel suffix (e.g. 'my-dest@provider_subaccount'). "
+            f"Unknown level {level_hint!r}. Valid values: {valid}"
+        )
+    if level is not None and level != parsed_level:
+        raise ValueError(
+            f"Conflicting levels: name contains '@{level_hint}' but "
+            f"level={level!r} was also passed explicitly. "
+            f"Use only the level= parameter instead."
+        )
+    return base_name, parsed_level
+
+
 class DestinationClient:
     """Client for SAP Destination Service operations.
 
@@ -240,6 +297,7 @@ def get_instance_destination(
             DeprecationWarning,
             stacklevel=2,
         )
+        validate_resource_name(name)
         try:
             if self._should_use_proxy(proxy_enabled):
                 return TransparentProxyDestination.from_proxy(
@@ -295,6 +353,7 @@ def get_subaccount_destination(
             DeprecationWarning,
             stacklevel=2,
         )
+        validate_resource_name(name)
         try:
             if self._should_use_proxy(proxy_enabled) and self._transparent_proxy:
                 return TransparentProxyDestination.from_proxy(
@@ -386,6 +445,8 @@ def get_destination(
             dest = client.get_destination("my-api", proxy_enabled=True)
             ```
         """
+        name, level = _split_name_and_level(name, level)
+        validate_resource_name(name)
         try:
             if self._should_use_proxy(proxy_enabled):
                 return TransparentProxyDestination.from_proxy(
@@ -434,9 +495,9 @@ def get_destination(
 
             # Build path with optional level hint
             path = (
-                f"{API_V2}/destinations/{name}@{level.value}"
+                f"{API_V2}/destinations/{encode_path_segment(name)}@{level.value}"
                 if level
-                else f"{API_V2}/destinations/{name}"
+                else f"{API_V2}/destinations/{encode_path_segment(name)}"
             )
 
             params: Dict[str, Any] = {}
@@ -501,6 +562,7 @@ def create_destination(
             DestinationOperationError: For unexpected errors.
         """
         coll = self._sub_path_for_level(level)
+        validate_resource_name(dest.name)
         body = dest.to_dict()
 
         try:
@@ -542,6 +604,7 @@ def update_destination(
             DestinationOperationError: For unexpected errors.
         """
         coll = self._sub_path_for_level(level)
+        validate_resource_name(dest.name)
         body = dest.to_dict()
 
         try:
@@ -579,12 +642,13 @@ def delete_destination(
             DestinationOperationError: For unexpected errors.
         """
         coll = self._sub_path_for_level(level)
+        validate_resource_name(name)
 
         try:
             _request(
                 self._http,
                 HttpMethod.DELETE,
-                f"{API_V1}/{coll}/{name}",
+                f"{API_V1}/{coll}/{encode_path_segment(name)}",
                 tenant_subdomain=tenant,
             )
         except HttpError:
@@ -616,12 +680,13 @@ def get_destination_labels(
         Raises:
             DestinationOperationError: If an HTTP error occurs or response parsing fails.
         """
+        validate_resource_name(name)
         try:
             path = self._sub_path_for_level(level)
             resp = _request(
                 self._http,
                 HttpMethod.GET,
-                f"{API_V1}/{path}/{name}/labels",
+                f"{API_V1}/{path}/{encode_path_segment(name)}/labels",
                 tenant_subdomain=tenant,
             )
             data = resp.json()
@@ -660,12 +725,13 @@ def update_destination_labels(
             DestinationOperationError: For unexpected errors.
         """
         resolved_level = level or Level.SUB_ACCOUNT
+        validate_resource_name(name)
         try:
             path = self._sub_path_for_level(resolved_level)
             _request(
                 self._http,
                 HttpMethod.PUT,
-                f"{API_V1}/{path}/{name}/labels",
+                f"{API_V1}/{path}/{encode_path_segment(name)}/labels",
                 json=[lbl.to_dict() for lbl in labels],
                 tenant_subdomain=tenant,
             )
@@ -697,12 +763,13 @@ def patch_destination_labels(
             DestinationOperationError: For unexpected errors.
         """
         resolved_level = level or Level.SUB_ACCOUNT
+        validate_resource_name(name)
         try:
             path = self._sub_path_for_level(resolved_level)
             _request(
                 self._http,
                 HttpMethod.PATCH,
-                f"{API_V1}/{path}/{name}/labels",
+                f"{API_V1}/{path}/{encode_path_segment(name)}/labels",
                 json=patch.to_dict(),
                 tenant_subdomain=tenant,
             )
@@ -769,7 +836,7 @@ def _get_destination(
             resp = _request(
                 self._http,
                 HttpMethod.GET,
-                f"{API_V1}/{path}/{name}",
+                f"{API_V1}/{path}/{encode_path_segment(name)}",
                 tenant_subdomain=tenant_subdomain,
             )
             data = resp.json()