fix: reject unsafe URL path segments in the 4.x SDK

tests/test_mfa.py

@@ -172,6 +172,18 @@ def test_delete_factor_no_id(self):
             self.mfa.delete_factor(authentication_factor_id=None)
         assert "Incomplete arguments. Need to specify a factor ID." in str(err.value)
 
+    @pytest.mark.parametrize("segment", ["", ".", ".."])
+    def test_delete_factor_rejects_invalid_path_segment_before_request(
+        self, segment, capture_and_mock_request
+    ):
+        request_args, request_kwargs = capture_and_mock_request("delete", None, 204)
+
+        with pytest.raises(ValueError):
+            self.mfa.delete_factor(segment)
+
+        assert request_args == []
+        assert request_kwargs == {}
+
     def test_delete_factor_success(self, mock_request_method):
         mock_request_method("delete", None, 200)
         response = self.mfa.delete_factor("auth_factor_01FZ4TS14D1PHFNZ9GF6YD8M1F")

tests/test_user_management.py

@@ -253,6 +253,32 @@ def mock_invitations(self):
         }
         return dict_response
 
+    def test_get_user_encodes_path_segment(self, mock_user, capture_and_mock_request):
+        url, _ = capture_and_mock_request("get", mock_user, 200)
+
+        self.user_management.get_user("a/b")
+
+        assert url[0] == workos.base_api_url + "user_management/users/a%2Fb"
+
+    @pytest.mark.parametrize("segment", ["", ".", ".."])
+    @pytest.mark.parametrize(
+        "method, http_method",
+        [
+            ("delete_user", "delete"),
+            ("deactivate_organization_membership", "put"),
+        ],
+    )
+    def test_rejects_invalid_path_segment_before_request(
+        self, segment, method, http_method, capture_and_mock_request
+    ):
+        request_args, request_kwargs = capture_and_mock_request(http_method, None, 204)
+
+        with pytest.raises(ValueError):
+            getattr(self.user_management, method)(segment)
+
+        assert request_args == []
+        assert request_kwargs == {}
+
     def test_get_user(self, mock_user, capture_and_mock_request):
         url, request_kwargs = capture_and_mock_request("get", mock_user, 200)
 

tests/utils/test_requests.py

@@ -6,7 +6,7 @@
     BadRequestException,
     ServerException,
 )
-from workos.utils.request import RequestHelper, BASE_HEADERS
+from workos.utils.request import RequestHelper, BASE_HEADERS, encode_path_segment
 
 STATUS_CODE_TO_EXCEPTION_MAPPING = {
     400: BadRequestException,
@@ -16,6 +16,30 @@
 }
 
 
+@pytest.mark.parametrize("segment", ["", ".", ".."])
+def test_encode_path_segment_rejects_invalid_segments(segment):
+    with pytest.raises(ValueError) as err:
+        encode_path_segment(segment)
+
+    assert str(err.value) == "Path segments must not be empty, '.' or '..'."
+
+
+@pytest.mark.parametrize(
+    "value, expected",
+    [
+        ("user_123", "user_123"),
+        (123, "123"),
+        ("a/b", "a%2Fb"),
+        ("a/../b", "a%2F..%2Fb"),
+        ("x?y", "x%3Fy"),
+        ("x#y", "x%23y"),
+        ("%2e%2e", "%252e%252e"),
+    ],
+)
+def test_encode_path_segment(value, expected):
+    assert encode_path_segment(value) == expected
+
+
 class TestRequestHelper(object):
     def test_set_base_api_url(self):
         pass
@@ -150,6 +174,17 @@ def test_request_parses_json_when_encoding_in_content_type(
 
         assert RequestHelper().request("ok_place") == {"foo": "bar"}
 
+    @pytest.mark.parametrize("segment", ["", ".", ".."])
+    def test_build_url_rejects_invalid_path_segment(self, segment):
+        with pytest.raises(ValueError):
+            RequestHelper().build_parameterized_url("users/{user}/x", user=segment)
+
+    def test_build_url_encodes_path_segment(self):
+        assert (
+            RequestHelper().build_parameterized_url("users/{user}", user="a/../b")
+            == "users/a%2F..%2Fb"
+        )
+
     def test_build_url(self):
         assert RequestHelper().build_parameterized_url("a/b/c") == "a/b/c"
         assert RequestHelper().build_parameterized_url("a/{b}/c", b="b") == "a/b/c"

workos/audit_logs.py

@@ -1,7 +1,12 @@
 from warnings import warn
 import workos
 from workos.resources.audit_logs_export import WorkOSAuditLogExport
-from workos.utils.request import RequestHelper, REQUEST_METHOD_GET, REQUEST_METHOD_POST
+from workos.utils.request import (
+    encode_path_segment,
+    RequestHelper,
+    REQUEST_METHOD_GET,
+    REQUEST_METHOD_POST,
+)
 from workos.utils.validation import AUDIT_LOGS_MODULE, validate_settings
 
 EVENTS_PATH = "audit_logs/events"
@@ -126,7 +131,7 @@ def get_export(self, export_id):
         """
 
         response = self.request_helper.request(
-            "{0}/{1}".format(EXPORTS_PATH, export_id),
+            "{0}/{1}".format(EXPORTS_PATH, encode_path_segment(export_id)),
             method=REQUEST_METHOD_GET,
             token=workos.api_key,
         )

workos/directory_sync.py

@@ -2,6 +2,7 @@
 import workos
 from workos.utils.pagination_order import Order
 from workos.utils.request import (
+    encode_path_segment,
     RequestHelper,
     REQUEST_METHOD_DELETE,
     REQUEST_METHOD_GET,
@@ -322,7 +323,7 @@ def get_user(self, user):
             dict: Directory User response from WorkOS.
         """
         response = self.request_helper.request(
-            "directory_users/{user}".format(user=user),
+            "directory_users/{user}".format(user=encode_path_segment(user)),
             method=REQUEST_METHOD_GET,
             token=workos.api_key,
         )
@@ -339,7 +340,7 @@ def get_group(self, group):
             dict: Directory Group response from WorkOS.
         """
         response = self.request_helper.request(
-            "directory_groups/{group}".format(group=group),
+            "directory_groups/{group}".format(group=encode_path_segment(group)),
             method=REQUEST_METHOD_GET,
             token=workos.api_key,
         )
@@ -358,7 +359,7 @@ def get_directory(self, directory):
         """
 
         response = self.request_helper.request(
-            "directories/{directory}".format(directory=directory),
+            "directories/{directory}".format(directory=encode_path_segment(directory)),
             method=REQUEST_METHOD_GET,
             token=workos.api_key,
         )
@@ -517,7 +518,7 @@ def get_directory(self, directory):
         """
 
         response = self.request_helper.request(
-            "directories/{directory}".format(directory=directory),
+            "directories/{directory}".format(directory=encode_path_segment(directory)),
             method=REQUEST_METHOD_GET,
             token=workos.api_key,
         )
@@ -534,7 +535,7 @@ def delete_directory(self, directory):
             dict: Directories response from WorkOS.
         """
         return self.request_helper.request(
-            "directories/{directory}".format(directory=directory),
+            "directories/{directory}".format(directory=encode_path_segment(directory)),
             method=REQUEST_METHOD_DELETE,
             token=workos.api_key,
         )

workos/organizations.py

@@ -2,6 +2,7 @@
 import workos
 from workos.utils.pagination_order import Order
 from workos.utils.request import (
+    encode_path_segment,
     RequestHelper,
     REQUEST_METHOD_DELETE,
     REQUEST_METHOD_GET,
@@ -162,7 +163,9 @@ def get_organization(self, organization):
             dict: Organization response from WorkOS
         """
         response = self.request_helper.request(
-            "organizations/{organization}".format(organization=organization),
+            "organizations/{organization}".format(
+                organization=encode_path_segment(organization)
+            ),
             method=REQUEST_METHOD_GET,
             token=workos.api_key,
         )
@@ -177,7 +180,9 @@ def get_organization_by_lookup_key(self, lookup_key):
             dict: Organization response from WorkOS
         """
         response = self.request_helper.request(
-            "organizations/by_lookup_key/{lookup_key}".format(lookup_key=lookup_key),
+            "organizations/by_lookup_key/{lookup_key}".format(
+                lookup_key=encode_path_segment(lookup_key)
+            ),
             method=REQUEST_METHOD_GET,
             token=workos.api_key,
         )
@@ -282,7 +287,9 @@ def update_organization(
             params["lookup_key"] = lookup_key
 
         response = self.request_helper.request(
-            "organizations/{organization}".format(organization=organization),
+            "organizations/{organization}".format(
+                organization=encode_path_segment(organization)
+            ),
             method=REQUEST_METHOD_PUT,
             params=params,
             token=workos.api_key,
@@ -297,7 +304,9 @@ def delete_organization(self, organization):
             organization (str): Organization unique identifier
         """
         return self.request_helper.request(
-            "organizations/{organization}".format(organization=organization),
+            "organizations/{organization}".format(
+                organization=encode_path_segment(organization)
+            ),
             method=REQUEST_METHOD_DELETE,
             token=workos.api_key,
         )

workos/passwordless.py

@@ -1,5 +1,9 @@
 import workos
-from workos.utils.request import RequestHelper, REQUEST_METHOD_POST
+from workos.utils.request import (
+    encode_path_segment,
+    RequestHelper,
+    REQUEST_METHOD_POST,
+)
 from workos.utils.validation import PASSWORDLESS_MODULE, validate_settings
 from workos.resources.passwordless import WorkOSPasswordlessSession
 
@@ -60,7 +64,9 @@ def send_session(self, session_id):
             boolean: Returns True
         """
         self.request_helper.request(
-            "passwordless/sessions/{session_id}/send".format(session_id=session_id),
+            "passwordless/sessions/{session_id}/send".format(
+                session_id=encode_path_segment(session_id)
+            ),
             method=REQUEST_METHOD_POST,
             token=workos.api_key,
         )

workos/sso.py

@@ -10,6 +10,7 @@
 from workos.utils.connection_types import ConnectionType
 from workos.utils.sso_provider_types import SsoProviderType
 from workos.utils.request import (
+    encode_path_segment,
     RequestHelper,
     RESPONSE_TYPE_CODE,
     REQUEST_METHOD_DELETE,
@@ -207,7 +208,9 @@ def get_connection(self, connection):
             dict: Connection response from WorkOS.
         """
         response = self.request_helper.request(
-            "connections/{connection}".format(connection=connection),
+            "connections/{connection}".format(
+                connection=encode_path_segment(connection)
+            ),
             method=REQUEST_METHOD_GET,
             token=workos.api_key,
         )
@@ -391,7 +394,9 @@ def delete_connection(self, connection):
             connection (str): Connection unique identifier
         """
         return self.request_helper.request(
-            "connections/{connection}".format(connection=connection),
+            "connections/{connection}".format(
+                connection=encode_path_segment(connection)
+            ),
             method=REQUEST_METHOD_DELETE,
             token=workos.api_key,
         )