From fe604977d51d815f15be0892791fd9b6f79cf6a8 Mon Sep 17 00:00:00 2001 From: "Garen J. Torikian" Date: Fri, 11 Sep 2026 12:05:34 -0400 Subject: [PATCH] fix: reject unsafe URL path segments Dot segments can be normalized into a different endpoint, making an identifier capable of retargeting a destructive request. URL quoting alone cannot prevent this because '.' and '..' remain unescaped. Empty identifiers and bare dot segments now raise ValueError before any HTTP request. Slashes, question marks, hashes, and percent signs in path identifiers are encoded rather than interpreted as URL structure. Callers must pass raw identifiers; pre-encoded values are double-encoded. This applies the VULN-1272 remediation to the requests-based 4.x SDK. --- tests/test_mfa.py | 12 +++++++++ tests/test_user_management.py | 26 ++++++++++++++++++ tests/utils/test_requests.py | 37 ++++++++++++++++++++++++- workos/audit_logs.py | 9 +++++-- workos/directory_sync.py | 11 ++++---- workos/organizations.py | 17 +++++++++--- workos/passwordless.py | 10 +++++-- workos/sso.py | 9 +++++-- workos/user_management.py | 51 ++++++++++++++++++++++------------- workos/utils/request.py | 14 +++++++++- 10 files changed, 161 insertions(+), 35 deletions(-) diff --git a/tests/test_mfa.py b/tests/test_mfa.py index d34f5298..4831c64d 100644 --- a/tests/test_mfa.py +++ b/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") diff --git a/tests/test_user_management.py b/tests/test_user_management.py index cd81834f..6c1105c7 100644 --- a/tests/test_user_management.py +++ b/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) diff --git a/tests/utils/test_requests.py b/tests/utils/test_requests.py index fea03e0f..e69e6848 100644 --- a/tests/utils/test_requests.py +++ b/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" diff --git a/workos/audit_logs.py b/workos/audit_logs.py index ec91d2d1..cd53d75a 100644 --- a/workos/audit_logs.py +++ b/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, ) diff --git a/workos/directory_sync.py b/workos/directory_sync.py index 2a1dd43a..0f344cad 100644 --- a/workos/directory_sync.py +++ b/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, ) diff --git a/workos/organizations.py b/workos/organizations.py index 83e706dc..2e3763f6 100644 --- a/workos/organizations.py +++ b/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, ) diff --git a/workos/passwordless.py b/workos/passwordless.py index 008c471a..95cf14da 100644 --- a/workos/passwordless.py +++ b/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, ) diff --git a/workos/sso.py b/workos/sso.py index d26cb2e8..88f7e82a 100644 --- a/workos/sso.py +++ b/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, ) diff --git a/workos/user_management.py b/workos/user_management.py index 09f5dbfb..fbac20cd 100644 --- a/workos/user_management.py +++ b/workos/user_management.py @@ -17,6 +17,7 @@ from workos.utils.pagination_order import Order from workos.utils.um_provider_types import UserManagementProviderType from workos.utils.request import ( + encode_path_segment, RequestHelper, RESPONSE_TYPE_CODE, REQUEST_METHOD_POST, @@ -81,7 +82,7 @@ def get_user(self, user_id): headers = {} response = self.request_helper.request( - USER_DETAIL_PATH.format(user_id), + USER_DETAIL_PATH.format(encode_path_segment(user_id)), method=REQUEST_METHOD_GET, headers=headers, token=workos.api_key, @@ -200,7 +201,7 @@ def update_user(self, user_id, payload): dict: Updated User response from WorkOS. """ response = self.request_helper.request( - USER_DETAIL_PATH.format(user_id), + USER_DETAIL_PATH.format(encode_path_segment(user_id)), method=REQUEST_METHOD_PUT, params=payload, token=workos.api_key, @@ -215,7 +216,7 @@ def delete_user(self, user_id): user_id (str) - User unique identifier """ self.request_helper.request( - USER_DETAIL_PATH.format(user_id), + USER_DETAIL_PATH.format(encode_path_segment(user_id)), method=REQUEST_METHOD_DELETE, token=workos.api_key, ) @@ -270,7 +271,9 @@ def update_organization_membership( } response = self.request_helper.request( - ORGANIZATION_MEMBERSHIP_DETAIL_PATH.format(organization_membership_id), + ORGANIZATION_MEMBERSHIP_DETAIL_PATH.format( + encode_path_segment(organization_membership_id) + ), method=REQUEST_METHOD_PUT, params=params, headers=headers, @@ -290,7 +293,9 @@ def get_organization_membership(self, organization_membership_id): headers = {} response = self.request_helper.request( - ORGANIZATION_MEMBERSHIP_DETAIL_PATH.format(organization_membership_id), + ORGANIZATION_MEMBERSHIP_DETAIL_PATH.format( + encode_path_segment(organization_membership_id) + ), method=REQUEST_METHOD_GET, headers=headers, token=workos.api_key, @@ -377,7 +382,9 @@ def delete_organization_membership(self, organization_membership_id): organization_membership_id (str) - The unique ID of the Organization Membership. """ self.request_helper.request( - ORGANIZATION_MEMBERSHIP_DETAIL_PATH.format(organization_membership_id), + ORGANIZATION_MEMBERSHIP_DETAIL_PATH.format( + encode_path_segment(organization_membership_id) + ), method=REQUEST_METHOD_DELETE, token=workos.api_key, ) @@ -391,7 +398,9 @@ def deactivate_organization_membership(self, organization_membership_id): dict: OrganizationMembership response from WorkOS. """ response = self.request_helper.request( - ORGANIZATION_MEMBERSHIP_DEACTIVATE_PATH.format(organization_membership_id), + ORGANIZATION_MEMBERSHIP_DEACTIVATE_PATH.format( + encode_path_segment(organization_membership_id) + ), method=REQUEST_METHOD_PUT, token=workos.api_key, ) @@ -407,7 +416,9 @@ def reactivate_organization_membership(self, organization_membership_id): dict: OrganizationMembership response from WorkOS. """ response = self.request_helper.request( - ORGANIZATION_MEMBERSHIP_REACTIVATE_PATH.format(organization_membership_id), + ORGANIZATION_MEMBERSHIP_REACTIVATE_PATH.format( + encode_path_segment(organization_membership_id) + ), method=REQUEST_METHOD_PUT, token=workos.api_key, ) @@ -863,7 +874,7 @@ def get_password_reset(self, password_reset_id): headers = {} response = self.request_helper.request( - PASSWORD_RESET_DETAIL_PATH.format(password_reset_id), + PASSWORD_RESET_DETAIL_PATH.format(encode_path_segment(password_reset_id)), method=REQUEST_METHOD_GET, headers=headers, token=workos.api_key, @@ -977,7 +988,9 @@ def get_email_verification(self, email_verification_id): headers = {} response = self.request_helper.request( - EMAIL_VERIFICATION_DETAIL_PATH.format(email_verification_id), + EMAIL_VERIFICATION_DETAIL_PATH.format( + encode_path_segment(email_verification_id) + ), method=REQUEST_METHOD_GET, headers=headers, token=workos.api_key, @@ -1001,7 +1014,7 @@ def send_verification_email( headers = {} response = self.request_helper.request( - USER_SEND_VERIFICATION_EMAIL_PATH.format(user_id), + USER_SEND_VERIFICATION_EMAIL_PATH.format(encode_path_segment(user_id)), method=REQUEST_METHOD_POST, headers=headers, token=workos.api_key, @@ -1032,7 +1045,7 @@ def verify_email( } response = self.request_helper.request( - USER_VERIFY_EMAIL_CODE_PATH.format(user_id), + USER_VERIFY_EMAIL_CODE_PATH.format(encode_path_segment(user_id)), method=REQUEST_METHOD_POST, headers=headers, params=payload, @@ -1053,7 +1066,7 @@ def get_magic_auth(self, magic_auth_id): headers = {} response = self.request_helper.request( - MAGIC_AUTH_DETAIL_PATH.format(magic_auth_id), + MAGIC_AUTH_DETAIL_PATH.format(encode_path_segment(magic_auth_id)), method=REQUEST_METHOD_GET, headers=headers, token=workos.api_key, @@ -1156,7 +1169,7 @@ def enroll_auth_factor( } response = self.request_helper.request( - USER_AUTH_FACTORS_PATH.format(user_id), + USER_AUTH_FACTORS_PATH.format(encode_path_segment(user_id)), method=REQUEST_METHOD_POST, headers=headers, params=payload, @@ -1191,7 +1204,7 @@ def list_auth_factors( dict: List of Authentication Factors for a User from WorkOS. """ response = self.request_helper.request( - USER_AUTH_FACTORS_PATH.format(user_id), + USER_AUTH_FACTORS_PATH.format(encode_path_segment(user_id)), method=REQUEST_METHOD_GET, token=workos.api_key, ) @@ -1217,7 +1230,7 @@ def get_invitation(self, invitation_id): headers = {} response = self.request_helper.request( - INVITATION_DETAIL_PATH.format(invitation_id), + INVITATION_DETAIL_PATH.format(encode_path_segment(invitation_id)), method=REQUEST_METHOD_GET, headers=headers, token=workos.api_key, @@ -1237,7 +1250,9 @@ def find_invitation_by_token(self, invitation_token): headers = {} response = self.request_helper.request( - INVITATION_DETAIL_BY_TOKEN_PATH.format(invitation_token), + INVITATION_DETAIL_BY_TOKEN_PATH.format( + encode_path_segment(invitation_token) + ), method=REQUEST_METHOD_GET, headers=headers, token=workos.api_key, @@ -1363,7 +1378,7 @@ def revoke_invitation(self, invitation_id): headers = {} response = self.request_helper.request( - INVITATION_REVOKE_PATH.format(invitation_id), + INVITATION_REVOKE_PATH.format(encode_path_segment(invitation_id)), method=REQUEST_METHOD_POST, headers=headers, token=workos.api_key, diff --git a/workos/utils/request.py b/workos/utils/request.py index c5d14bb2..98255eca 100644 --- a/workos/utils/request.py +++ b/workos/utils/request.py @@ -27,6 +27,18 @@ REQUEST_METHOD_PUT = "put" +def encode_path_segment(value): + """Encode a raw (not pre-encoded) value for use as one URL path segment. + + Empty and dot segments raise ValueError to prevent request retargeting. + Pre-encoded values will be double-encoded. + """ + segment = str(value) + if segment in ("", ".", ".."): + raise ValueError("Path segments must not be empty, '.' or '..'.") + return urllib.parse.quote(segment, safe="") + + class RequestHelper(object): def __init__(self): self.set_base_api_url(workos.base_api_url) @@ -47,7 +59,7 @@ def generate_api_url(self, path): return self.base_api_url.format(path) def build_parameterized_url(self, url, **params): - escaped_params = {k: urllib.parse.quote(str(v)) for k, v in params.items()} + escaped_params = {k: encode_path_segment(v) for k, v in params.items()} return url.format(**escaped_params) def request(