Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 21 additions & 6 deletions src/sap_cloud_sdk/print/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.

Expand Down Expand Up @@ -160,11 +172,12 @@ def create_queue(self, queue: PrintQueue) -> None:
the path parameter — this is enforced automatically.

Raises:
PrintOperationError: If the request fails.
PrintOperationError: If the request fails or the queue name is '.' or '..'.
UnicodeEncodeError: If the queue name cannot be encoded as UTF-8.
"""
try:
self.put(
f"{_QUEUES_PATH}/{queue.qname}",
f"{_QUEUES_PATH}/{_encode_path_segment(queue.qname)}",
json=queue.to_dict(),
headers=_IF_NONE_MATCH,
)
Expand All @@ -188,10 +201,11 @@ def get_print_profiles(self, qname: str) -> list[PrintProfile]:
List of PrintProfile objects for the queue.

Raises:
PrintOperationError: If the request fails or the response cannot be parsed.
PrintOperationError: If the request fails, the response cannot be parsed,
or the queue name is '.', '..', or cannot be encoded as UTF-8.
"""
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:
Expand Down Expand Up @@ -255,13 +269,14 @@ def create_print_task(self, task: PrintTask) -> None:
task: PrintTask describing the print job.

Raises:
PrintOperationError: If the request fails.
PrintOperationError: If the request fails or the task ID is '.' or '..'.
UnicodeEncodeError: If the task ID cannot be encoded as UTF-8.
"""
if not task.username:
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,
)
Expand Down
3 changes: 3 additions & 0 deletions src/sap_cloud_sdk/print/user-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,9 @@ except HttpError as e:
| `PrintOperationError` | An API operation fails (wraps `HttpError`) |
| `HttpError` | Raw HTTP error with `status_code` and `response_text` |

The SDK automatically URL-encodes `qname` and `item_id` when building URL paths.
Literal `.` or `..` identifiers raise `PrintOperationError` before the Print Service request.

---

## Models
Expand Down
103 changes: 103 additions & 0 deletions tests/print/unit/test_url_identifiers.py
Original file line number Diff line number Diff line change
@@ -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
Loading