diff --git a/resend/_path.py b/resend/_path.py new file mode 100644 index 0000000..c077e4f --- /dev/null +++ b/resend/_path.py @@ -0,0 +1,18 @@ +import re + +_DOT_SEGMENT = re.compile(r"^(?:\.|%2e){1,2}$", re.IGNORECASE) + + +def has_dot_segment(path: str) -> bool: + """Whether a request path contains a `.` or `..` segment. + + HTTP clients resolve these before sending (requests/urllib3 and httpx both + do), so an id such as "../api-keys/key_id" would reach another endpoint. + """ + pathname = re.split(r"[?#]", path, maxsplit=1)[0] + return any(_DOT_SEGMENT.match(segment) for segment in pathname.split("/")) + + +def ensure_no_dot_segments(path: str) -> None: + if has_dot_segment(path): + raise ValueError("Path parameters cannot be `.` or `..`.") diff --git a/resend/async_request.py b/resend/async_request.py index be17658..e791e65 100644 --- a/resend/async_request.py +++ b/resend/async_request.py @@ -4,6 +4,7 @@ from typing_extensions import Literal, TypeVar import resend +from resend._path import ensure_no_dot_segments from resend.exceptions import (NoContentError, ResendError, raise_for_code_and_type) from resend.http_client_async import AsyncHTTPClient @@ -37,6 +38,7 @@ def __init__( self._response_status_code: Optional[int] = None async def perform(self) -> Union[T, None]: + ensure_no_dot_segments(self.path) data = await self.make_request(url=f"{resend.api_url}{self.path}") body_status_code = data.get("statusCode") if isinstance(data, dict) else None diff --git a/resend/request.py b/resend/request.py index 56d27c1..92f1d65 100644 --- a/resend/request.py +++ b/resend/request.py @@ -4,6 +4,7 @@ from typing_extensions import Literal, TypeVar import resend +from resend._path import ensure_no_dot_segments from resend.exceptions import (NoContentError, ResendError, raise_for_code_and_type) from resend.response import ResponseDict @@ -36,6 +37,7 @@ def __init__( self._response_status_code: Optional[int] = None def perform(self) -> Union[T, None]: + ensure_no_dot_segments(self.path) data = self.make_request(url=f"{resend.api_url}{self.path}") body_status_code = data.get("statusCode") if isinstance(data, dict) else None diff --git a/tests/dot_segment_path_test.py b/tests/dot_segment_path_test.py new file mode 100644 index 0000000..5643164 --- /dev/null +++ b/tests/dot_segment_path_test.py @@ -0,0 +1,96 @@ +from typing import Any, Callable, cast +from unittest.mock import AsyncMock, create_autospec + +import pytest + +import resend +from resend._path import has_dot_segment +from resend.http_client import HTTPClient +from resend.http_client_async import AsyncHTTPClient + +# flake8: noqa + +DOT_SEGMENT_CALLS = [ + lambda: resend.Emails.get(".."), + lambda: resend.Emails.cancel(".."), + lambda: resend.Domains.remove("."), + lambda: resend.Domains.remove("../api-keys/key_id"), + lambda: resend.Domains.get("%2E%2E"), +] + + +@pytest.fixture +def mock_client() -> Any: + resend.api_key = "re_test" + previous = resend.default_http_client + client = create_autospec(HTTPClient, instance=True) + client.request.return_value = (b"{}", 200, {"Content-Type": "application/json"}) + resend.default_http_client = client + yield client + resend.default_http_client = previous + + +@pytest.mark.parametrize("call", DOT_SEGMENT_CALLS) +def test_dot_segment_ids_are_rejected_before_sending( + mock_client: Any, call: Callable[[], Any] +) -> None: + with pytest.raises(ValueError, match="cannot be `.` or `..`"): + call() + mock_client.request.assert_not_called() + + +def test_dotted_ids_are_still_sent(mock_client: Any) -> None: + resend.Domains.get("example.com") + resend.Contacts.get(email="john.doe@example.com") + resend.Domains.get("...") + urls = [c.kwargs["url"] for c in mock_client.request.call_args_list] + assert urls[0].endswith("/domains/example.com") + assert urls[1].endswith("/contacts/john.doe%40example.com") + assert urls[2].endswith("/domains/...") + + +@pytest.mark.asyncio +async def test_dot_segment_ids_are_rejected_before_sending_async() -> None: + resend.api_key = "re_test" + previous = resend.default_async_http_client + client = create_autospec(AsyncHTTPClient, instance=True) + client.request = AsyncMock( + return_value=(b"{}", 200, {"Content-Type": "application/json"}) + ) + resend.default_async_http_client = cast(AsyncHTTPClient, client) + try: + with pytest.raises(ValueError, match="cannot be `.` or `..`"): + await resend.Domains.remove_async("../api-keys/key_id") + client.request.assert_not_called() + finally: + resend.default_async_http_client = previous + + +@pytest.mark.parametrize( + "path", + [ + "/emails/..", + "/emails/./cancel", + "/domains/../api-keys/key_id", + "/emails/%2E%2E/cancel", + "/emails/%2e.", + "/emails/.%2E?limit=1", + ], +) +def test_has_dot_segment(path: str) -> None: + assert has_dot_segment(path) + + +@pytest.mark.parametrize( + "path", + [ + "/emails/4ef9a417-02e9-4d39-ad75-9611e0fcc33c", + "/emails/...", + "/emails/..%2Fapi-keys", + "/contacts/john.doe@example.com", + "/templates?after=..", + "/emails/x#/..", + ], +) +def test_has_no_dot_segment(path: str) -> None: + assert not has_dot_segment(path)