Skip to content
Open
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
18 changes: 18 additions & 0 deletions resend/_path.py
Original file line number Diff line number Diff line change
@@ -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 `..`.")
2 changes: 2 additions & 0 deletions resend/async_request.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions resend/request.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
96 changes: 96 additions & 0 deletions tests/dot_segment_path_test.py
Original file line number Diff line number Diff line change
@@ -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)