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(