From d2c85fb24aa8f2464c3eafd794ad68074f0fc2da Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Pedro=20Brun?= Date: Thu, 5 Nov 2026 17:29:11 -0300 Subject: [PATCH] fix(print): encode queue and task identifiers in URL paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: João Pedro Brun --- src/sap_cloud_sdk/print/client.py | 18 +++- tests/print/unit/test_url_identifiers.py | 103 +++++++++++++++++++++++ 2 files changed, 118 insertions(+), 3 deletions(-) create mode 100644 tests/print/unit/test_url_identifiers.py diff --git a/src/sap_cloud_sdk/print/client.py b/src/sap_cloud_sdk/print/client.py index 1c7e1adb..258fd53d 100644 --- a/src/sap_cloud_sdk/print/client.py +++ b/src/sap_cloud_sdk/print/client.py @@ -6,6 +6,7 @@ import json import logging from typing import IO, Callable, Optional, Union +from urllib.parse import quote from requests.exceptions import RequestException @@ -24,6 +25,17 @@ logger = logging.getLogger(__name__) +def _encode_path_segment(value: object) -> str: + """Percent-encode a value as a single URL path segment. + + Literal '.' and '..' segments raise HttpError to prevent path traversal. + """ + segment = f"{value}" + if segment in (".", ".."): + raise HttpError("path identifier must not be '.' or '..'") + return quote(segment, safe="") + + def _resolve_username(access_token: str, fallback_client_id: str) -> str: """Decode the JWT access token and return the best available username. @@ -164,7 +176,7 @@ def create_queue(self, queue: PrintQueue) -> None: """ try: self.put( - f"{_QUEUES_PATH}/{queue.qname}", + f"{_QUEUES_PATH}/{_encode_path_segment(queue.qname)}", json=queue.to_dict(), headers=_IF_NONE_MATCH, ) @@ -191,7 +203,7 @@ def get_print_profiles(self, qname: str) -> list[PrintProfile]: PrintOperationError: If the request fails or the response cannot be parsed. """ try: - resp = self.get(f"{_QUEUES_PATH}/{qname}/profiles") + resp = self.get(f"{_QUEUES_PATH}/{_encode_path_segment(qname)}/profiles") data = resp.json() return [PrintProfile.from_dict(item) for item in data] except HttpError as e: @@ -261,7 +273,7 @@ def create_print_task(self, task: PrintTask) -> None: task.username = self.get_username() try: self.put( - f"{_TASKS_PATH}/{task.item_id}", + f"{_TASKS_PATH}/{_encode_path_segment(task.item_id)}", json=task.to_body(), headers=_IF_NONE_MATCH, ) diff --git a/tests/print/unit/test_url_identifiers.py b/tests/print/unit/test_url_identifiers.py new file mode 100644 index 00000000..06caf51e --- /dev/null +++ b/tests/print/unit/test_url_identifiers.py @@ -0,0 +1,103 @@ +"""Regression tests for Print identifier URL encoding.""" + +import json +from unittest.mock import Mock, patch + +import pytest +from requests import Response, Session + +from sap_cloud_sdk.core.protocol.http import HttpClient, XsuaaAuthProvider +from sap_cloud_sdk.print._models import PrintContent, PrintQueue, PrintTask +from sap_cloud_sdk.print.client import PrintClient +from sap_cloud_sdk.print.exceptions import HttpError, PrintOperationError + + +BASE_URL = "https://print.example.invalid" +OPERATIONS = [ + ("create_queue", "PUT", "/qm/api/v1/rest/queues/{identifier}"), + ("get_print_profiles", "GET", "/qm/api/v1/rest/queues/{identifier}/profiles"), + ("create_print_task", "PUT", "/qm/api/v1/rest/print-tasks/{identifier}"), +] + + +def _task(identifier): + return PrintTask( + item_id=identifier, + qname="queue", + username="user", + print_contents=[PrintContent(object_key=identifier, document_name="main.pdf")], + ) + + +def _invoke(client, operation, identifier): + if operation == "create_queue": + client.create_queue(PrintQueue(qname=identifier)) + elif operation == "get_print_profiles": + client.get_print_profiles(identifier) + else: + client.create_print_task(_task(identifier)) + + +@pytest.mark.parametrize("operation,method,path", OPERATIONS) +@pytest.mark.parametrize( + "identifier,encoded", + [ + ("../other?param=x#frag", "..%2Fother%3Fparam%3Dx%23frag"), + ("%2e%2e%2fother", "%252e%252e%252fother"), + ], +) +def test_prepared_urls_contain_identifiers_without_changing_bodies( + operation, method, path, identifier, encoded +): + with Session() as session: + auth = Mock(spec=XsuaaAuthProvider) + auth.base_url = BASE_URL + auth.get_session.return_value = session + client = PrintClient(HttpClient(BASE_URL, auth), auth, lambda: Mock()) + response = Response() + response.status_code = 200 + response._content = b"[]" + with patch.object(session, "send", return_value=response) as send: + _invoke(client, operation, identifier) + + prepared = send.call_args.args[0] + assert prepared.method == method + assert prepared.url == BASE_URL + path.format(identifier=encoded) + if operation != "get_print_profiles": + body = json.loads(prepared.body) + if operation == "create_queue": + assert body["qname"] == identifier + else: + assert body["printContents"][0]["objectKey"] == identifier + + +@pytest.mark.parametrize("operation,method,path", OPERATIONS) +@pytest.mark.parametrize("identifier", [".", ".."]) +def test_dot_segments_raise_existing_operation_error_without_http( + operation, method, path, identifier +): + http = Mock(spec=HttpClient) + client = PrintClient(http, Mock(), lambda: Mock()) + with pytest.raises(PrintOperationError) as error: + _invoke(client, operation, identifier) + assert isinstance(error.value.__cause__, HttpError) + http.request.assert_not_called() + + +def test_task_uuid_spelling_and_document_reference_are_preserved(): + identifier = "4056BB6C-F544-41D7-87E1-FFE818573E6E" + http = Mock(spec=HttpClient) + response = Response() + response.status_code = 204 + http.request.return_value = response + client = PrintClient(http, Mock(), lambda: Mock()) + task = _task(identifier) + + client.create_print_task(task) + + assert http.request.call_args.args[1] == f"/qm/api/v1/rest/print-tasks/{identifier}" + assert ( + http.request.call_args.kwargs["json"]["printContents"][0]["objectKey"] + == identifier + ) + assert task.item_id == identifier