fix(adms): escape OData V4 string keys to prevent injection

User-supplied IDs (job_id, document_type_id, etc.) were f-string
interpolated directly into single-quote-wrapped OData entity keys. A
value containing a single quote would break the URL or alter query
intent — e.g. an ID like ``x'); ...`` could in principle target a
different entity than the caller intended.

Add ``quote_odata_string_key`` helper (OData V4 §5.1.1.6.2 — single
quotes inside string literals must be doubled) and apply it at all 8
call sites in client.py (sync + async).

Also adds 4 unit tests covering simple values, embedded quotes,
injection neutralisation, and the empty string.

Author: adwitiyasushant

src/sap_cloud_sdk/adms/_http.py

@@ -32,6 +32,20 @@
 _CSRF_FETCH_VALUE = "Fetch"
 
 
+def quote_odata_string_key(value: str) -> str:
+    """Quote and escape a string value for use in an OData V4 entity key segment.
+
+    OData V4 §5.1.1.6.2 requires single-quoted string literals with embedded
+    single quotes doubled.  Without escaping, a value like ``O'Brien`` (or a
+    deliberately crafted ``'); ...``) breaks the URL or alters query intent.
+
+    Example::
+
+        path = f"Documents(DocID={quote_odata_string_key(doc_id)})"
+    """
+    return "'" + value.replace("'", "''") + "'"
+
+
 # ---------------------------------------------------------------------------
 # Sync HTTP wrapper
 # ---------------------------------------------------------------------------

src/sap_cloud_sdk/adms/client.py

@@ -30,7 +30,7 @@
 import httpx
 
 from sap_cloud_sdk.adms._auth import IasTokenFetcher
-from sap_cloud_sdk.adms._http import AdmsHttp, AsyncAdmsHttp
+from sap_cloud_sdk.adms._http import AdmsHttp, AsyncAdmsHttp, quote_odata_string_key
 from sap_cloud_sdk.adms._models import (
     AllowedDomain,
     BusinessObjectNodeType,
@@ -201,7 +201,7 @@ def get_download_url(
 
         fn_key = (
             f"{rel_key}/DownloadDocument("
-            f"DocContentVersionID='{doc_content_version_id}')"
+            f"DocContentVersionID={quote_odata_string_key(doc_content_version_id)})"
         )
         resp = self._http.get(fn_key, service_base=_SERVICE_PATH)
         return resp.json().get("value", "")
@@ -674,7 +674,7 @@ def create_document_type(self, payload: CreateDocumentTypeInput) -> DocumentType
     def delete_document_type(self, document_type_id: str) -> None:
         """Delete a document type classification."""
         self._http.delete(
-            f"DocumentType(DocumentTypeID='{document_type_id}')",
+            f"DocumentType(DocumentTypeID={quote_odata_string_key(document_type_id)})",
             service_base=_CONFIG_SERVICE_PATH,
         )
 
@@ -720,7 +720,7 @@ def delete_business_object_type(
     ) -> None:
         """Delete a business object node type registration."""
         self._http.delete(
-            f"BusinessObjectNodeType(BusinessObjectNodeTypeUniqueID='{business_object_node_type_unique_id}')",
+            f"BusinessObjectNodeType(BusinessObjectNodeTypeUniqueID={quote_odata_string_key(business_object_node_type_unique_id)})",
             service_base=_CONFIG_SERVICE_PATH,
         )
 
@@ -839,7 +839,7 @@ def get_status(
             Current :class:`~sap_cloud_sdk.adms._models.JobOutput`.
         """
         service = _ADMIN_SERVICE_PATH if use_admin_service else _SERVICE_PATH
-        path = f"JobStatus(JobID='{job_id}')"
+        path = f"JobStatus(JobID={quote_odata_string_key(job_id)})"
         resp = self._http.get(path, service_base=service)
         return JobOutput.from_dict(resp.json())
 
@@ -941,7 +941,7 @@ async def get_download_url(
 
         fn_key = (
             f"{rel_key}/DownloadDocument("
-            f"DocContentVersionID='{doc_content_version_id}')"
+            f"DocContentVersionID={quote_odata_string_key(doc_content_version_id)})"
         )
         resp = await self._http.get(fn_key, service_base=_SERVICE_PATH)
         return resp.json().get("value", "")
@@ -1316,7 +1316,7 @@ async def create_document_type(
     async def delete_document_type(self, document_type_id: str) -> None:
         """Async variant of :meth:`_ConfigurationApi.delete_document_type` — same semantics."""
         await self._http.delete(
-            f"DocumentType(DocumentTypeID='{document_type_id}')",
+            f"DocumentType(DocumentTypeID={quote_odata_string_key(document_type_id)})",
             service_base=_CONFIG_SERVICE_PATH,
         )
 
@@ -1362,7 +1362,7 @@ async def delete_business_object_type(
     ) -> None:
         """Async variant of :meth:`_ConfigurationApi.delete_business_object_type` — same semantics."""
         await self._http.delete(
-            f"BusinessObjectNodeType(BusinessObjectNodeTypeUniqueID='{business_object_node_type_unique_id}')",
+            f"BusinessObjectNodeType(BusinessObjectNodeTypeUniqueID={quote_odata_string_key(business_object_node_type_unique_id)})",
             service_base=_CONFIG_SERVICE_PATH,
         )
 
@@ -1471,7 +1471,7 @@ async def get_status(
             Current :class:`~sap_cloud_sdk.adms._models.JobOutput`.
         """
         service = _ADMIN_SERVICE_PATH if use_admin_service else _SERVICE_PATH
-        path = f"JobStatus(JobID='{job_id}')"
+        path = f"JobStatus(JobID={quote_odata_string_key(job_id)})"
         resp = await self._http.get(path, service_base=service)
         return JobOutput.from_dict(resp.json())
 

tests/adms/unit/test_http.py

@@ -7,7 +7,7 @@
 import requests
 
 from sap_cloud_sdk.adms._auth import IasTokenFetcher
-from sap_cloud_sdk.adms._http import AdmsHttp
+from sap_cloud_sdk.adms._http import AdmsHttp, quote_odata_string_key
 from sap_cloud_sdk.adms.config import AdmsConfig
 from sap_cloud_sdk.adms.exceptions import DocumentNotFoundError, HttpError
 
@@ -200,3 +200,23 @@ def test_service_jwt_uses_get_token(self, config, token_fetcher):
 
         token_fetcher.get_token.assert_called()
         token_fetcher.exchange_token.assert_not_called()
+
+
+class TestQuoteOdataStringKey:
+    def test_simple_value(self):
+        assert quote_odata_string_key("job-123") == "'job-123'"
+
+    def test_value_with_single_quote_is_doubled(self):
+        # OData V4 §5.1.1.6.2 — single quotes inside string literals must be doubled.
+        assert quote_odata_string_key("O'Brien") == "'O''Brien'"
+
+    def test_injection_attempt_is_neutralised(self):
+        # An attacker-controlled value must not break out of the quoted segment.
+        out = quote_odata_string_key("x'); DROP TABLE--")
+        assert out == "'x''); DROP TABLE--'"
+        # Result is one single-quoted literal, not two.
+        assert out.count("'") % 2 == 0
+
+    def test_empty_string(self):
+        assert quote_odata_string_key("") == "''"
+