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
24 changes: 22 additions & 2 deletions src/workos/_base_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,10 @@
RETRY_MULTIPLIER = 2


# Segments that URL normalization treats structurally; never valid identifiers.
_INVALID_PATH_SEGMENTS = frozenset({"", ".", ".."})


class _BaseWorkOSClient:
"""Shared WorkOS client implementation."""

Expand Down Expand Up @@ -145,7 +149,16 @@ def _encode_path(path: Sequence[str]) -> str:
Callers pass each path component as a separate element (e.g.
``("organizations", organization_id)``). Each element is URL-encoded
with ``safe=""`` so a caller-supplied id containing ``/``, ``?``,
``#``, ``%``, or ``..`` cannot escape its intended segment — this is
``#``, or ``%`` cannot escape its intended segment.

Percent-encoding alone does not cover dot segments: ``.`` is an
unreserved character that :func:`urllib.parse.quote` leaves as-is, and
httpx applies RFC 3986 dot-segment removal when it builds the request
URL, so a bare ``.`` or ``..`` segment would collapse the path onto the
parent resource (for example, turning a DELETE of one connected account
into a DELETE of the whole user). Empty, ``.`` and ``..`` segments are
never valid WorkOS identifiers, so they are rejected with
``ValueError`` before any request is made. Together these checks are
the structural protection against forged cross-resource API requests
under the application's API key.

Expand All @@ -157,7 +170,14 @@ def _encode_path(path: Sequence[str]) -> str:
raise TypeError(
"path must be a sequence of segments (e.g. a tuple), not a str"
)
return "/".join(quote(str(seg), safe="") for seg in path)
segments = [str(seg) for seg in path]
for seg in segments:
if seg in _INVALID_PATH_SEGMENTS:
raise ValueError(
f"invalid URL path segment {seg!r}: path segments must be "
"non-empty and cannot be '.' or '..'"
)
return "/".join(quote(seg, safe="") for seg in segments)

def _resolve_timeout(self, request_options: Optional[RequestOptions]) -> float:
timeout = self._request_timeout
Expand Down
69 changes: 69 additions & 0 deletions tests/test_generated_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -742,3 +742,72 @@ async def _sleep(_: float) -> None:
with pytest.raises(generated_client_module.WorkOSConnectionError):
await client.request("GET", ("test",))
await client.close()


class TestEncodePath:
"""Path segments are encoded so caller-supplied ids cannot retarget a request."""

@pytest.mark.parametrize("segment", ["", ".", ".."])
def test_rejects_empty_and_dot_segments(self, segment):
with pytest.raises(ValueError, match="invalid URL path segment"):
WorkOSClient._encode_path(
("user_management", "users", "user_01", "connected_accounts", segment)
)

@pytest.mark.parametrize("segment", ["", ".", ".."])
def test_async_client_rejects_empty_and_dot_segments(self, segment):
with pytest.raises(ValueError, match="invalid URL path segment"):
AsyncWorkOSClient._encode_path(("organizations", segment))

def test_rejects_bare_string_path(self):
with pytest.raises(TypeError):
WorkOSClient._encode_path("organizations/org_01") # type: ignore[arg-type]

@pytest.mark.parametrize(
("segment", "expected"),
[
("org_01", "org_01"),
("a/b", "a%2Fb"),
("a?b", "a%3Fb"),
("a#b", "a%23b"),
("a%b", "a%25b"),
("%2e%2e", "%252e%252e"),
("evil/../..", "evil%2F..%2F.."),
("...", "..."),
(".hidden", ".hidden"),
],
)
def test_structural_characters_stay_inside_their_segment(self, segment, expected):
assert (
WorkOSClient._encode_path(("organizations", segment))
== f"organizations/{expected}"
)

def test_dot_segment_is_rejected_before_any_request(self, httpx_mock):
client = WorkOSClient(api_key="sk_test_123", client_id="client_test")
try:
with pytest.raises(ValueError, match="invalid URL path segment"):
client.pipes.delete_user_connected_account("user_01ABC", "..")
finally:
client.close()
assert httpx_mock.get_requests() == []

def test_build_url_rejects_dot_segments(self):
client = WorkOSClient(api_key="sk_test_123", client_id="client_test")
try:
with pytest.raises(ValueError, match="invalid URL path segment"):
client.build_url(("sso", ".."), {"client_id": "client_test"})
finally:
client.close()


@pytest.mark.asyncio
class TestEncodePathAsync:
async def test_dot_segment_is_rejected_before_any_request(self, httpx_mock):
client = AsyncWorkOSClient(api_key="sk_test_123", client_id="client_test")
try:
with pytest.raises(ValueError, match="invalid URL path segment"):
await client.pipes.delete_user_connected_account("user_01ABC", "..")
finally:
await client.close()
assert httpx_mock.get_requests() == []
Loading