diff --git a/descope/management/_outbound_scim_base.py b/descope/management/_outbound_scim_base.py new file mode 100644 index 000000000..4a36d692e --- /dev/null +++ b/descope/management/_outbound_scim_base.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from typing import Any, Optional + + +class OutboundSCIMBase: + @staticmethod + def _compose_create_body( + app_id: str, + configuration: Optional[dict] = None, + ) -> dict: + body: dict[str, Any] = { + "appId": app_id, + } + if configuration is not None: + body["configuration"] = configuration + return body + + @staticmethod + def _compose_update_body( + app_id: str, + version: int, + configuration: Optional[dict] = None, + ) -> dict: + body: dict[str, Any] = { + "appId": app_id, + "version": version, + } + if configuration is not None: + body["configuration"] = configuration + return body diff --git a/descope/management/common.py b/descope/management/common.py index 9c2df5ea5..0a6466dcf 100644 --- a/descope/management/common.py +++ b/descope/management/common.py @@ -128,6 +128,13 @@ class MgmtV1: outbound_application_batch_upload_user_tokens_path = "/v1/mgmt/outbound/app/user/oauthtoken/batch/upload" outbound_application_batch_upload_tenant_tokens_path = "/v1/mgmt/outbound/app/tenant/oauthtoken/batch/upload" + # outbound scim + outbound_scim_create_path = "/v1/mgmt/outbound/scim/create" + outbound_scim_update_path = "/v1/mgmt/outbound/scim/update" + outbound_scim_delete_path = "/v1/mgmt/outbound/scim/delete" + outbound_scim_load_path = "/v1/mgmt/outbound/scim" + outbound_scim_set_enabled_path = "/v1/mgmt/outbound/scim/enabled/set" + # engine engine_create_path = "/v1/mgmt/engine/create" engine_update_path = "/v1/mgmt/engine/update" diff --git a/descope/management/outbound_scim.py b/descope/management/outbound_scim.py new file mode 100644 index 000000000..62841c4ba --- /dev/null +++ b/descope/management/outbound_scim.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +from typing import Optional + +from descope._http_base import HTTPBase +from descope.management._outbound_scim_base import OutboundSCIMBase +from descope.management.common import MgmtV1 + + +class OutboundSCIM(OutboundSCIMBase, HTTPBase): + def create_configuration( + self, + app_id: str, + configuration: Optional[dict] = None, + ) -> dict: + """ + Create a new outbound SCIM configuration on the federated SSO application + identified by app_id. The connector name is derived server-side from the app. + + Args: + app_id (str): The federated SSO application id this SCIM configuration binds to. + configuration (dict): Optional provider-specific SCIM configuration dictionary. + + Return value (dict): + Return dict in the format + {"configuration": {"appId": , "configuration": {...}, + "enabled": , "lastExportTime": , + "lastProcessingTime": , "failures": , + "version": }} + + Raise: + AuthException: raised if create operation fails + """ + response = self._http.post( + MgmtV1.outbound_scim_create_path, + body=OutboundSCIMBase._compose_create_body(app_id, configuration), + ) + return response.json() + + def update_configuration( + self, + app_id: str, + version: int, + configuration: Optional[dict] = None, + ) -> dict: + """ + Update the outbound SCIM configuration attached to the given federated SSO app. + The version must match the currently stored version — otherwise the update is + rejected as a conflict. + + Args: + app_id (str): The federated SSO application id. + version (int): The currently stored version, used for optimistic concurrency. + configuration (dict): Optional updated provider-specific SCIM configuration. + + Return value (dict): + Return dict in the format + {"configuration": {"appId": , ...}} + + Raise: + AuthException: raised if update operation fails + """ + response = self._http.post( + MgmtV1.outbound_scim_update_path, + body=OutboundSCIMBase._compose_update_body(app_id, version, configuration), + ) + return response.json() + + def delete_configuration( + self, + app_id: str, + ): + """ + Delete the outbound SCIM configuration attached to the given federated SSO app. + IMPORTANT: This action is irreversible. Use carefully. + + Args: + app_id (str): The federated SSO application id. + + Raise: + AuthException: raised if deletion operation fails + """ + self._http.post(MgmtV1.outbound_scim_delete_path, body={"appId": app_id}) + + def load_configuration( + self, + app_id: str, + ) -> dict: + """ + Load the outbound SCIM configuration attached to the given federated SSO app. + + Args: + app_id (str): The federated SSO application id. + + Return value (dict): + Return dict in the format + {"configuration": {"appId": , "configuration": {...}, + "enabled": , "lastExportTime": , + "lastProcessingTime": , "failures": , + "version": }} + + Raise: + AuthException: raised if load operation fails + """ + response = self._http.get(f"{MgmtV1.outbound_scim_load_path}/{app_id}") + return response.json() + + def set_enabled( + self, + app_id: str, + enabled: bool, + ) -> dict: + """ + Enable or disable the outbound SCIM configuration attached to the given + federated SSO app. + + Args: + app_id (str): The federated SSO application id. + enabled (bool): Whether the SCIM configuration should be enabled. + + Return value (dict): + Return dict in the format + {"configuration": {"appId": , "enabled": , ...}} + + Raise: + AuthException: raised if the operation fails + """ + response = self._http.post( + MgmtV1.outbound_scim_set_enabled_path, + body={"appId": app_id, "enabled": enabled}, + ) + return response.json() diff --git a/descope/management/outbound_scim_async.py b/descope/management/outbound_scim_async.py new file mode 100644 index 000000000..211bf19ba --- /dev/null +++ b/descope/management/outbound_scim_async.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from typing import Optional + +from descope._http_base import AsyncHTTPBase +from descope.management._outbound_scim_base import OutboundSCIMBase +from descope.management.common import MgmtV1 + + +class OutboundSCIMAsync(OutboundSCIMBase, AsyncHTTPBase): + """Async counterpart of OutboundSCIM — all HTTP calls are coroutines.""" + + async def create_configuration( + self, + app_id: str, + configuration: Optional[dict] = None, + ) -> dict: + """ + Create a new outbound SCIM configuration on the federated SSO application + identified by app_id. The connector name is derived server-side from the app. + + Args: + app_id (str): The federated SSO application id this SCIM configuration binds to. + configuration (dict): Optional provider-specific SCIM configuration dictionary. + + Return value (dict): + Return dict in the format + {"configuration": {"appId": , "configuration": {...}, + "enabled": , "lastExportTime": , + "lastProcessingTime": , "failures": , + "version": }} + + Raise: + AuthException: raised if create operation fails + """ + response = await self._http.post( + MgmtV1.outbound_scim_create_path, + body=OutboundSCIMBase._compose_create_body(app_id, configuration), + ) + return response.json() + + async def update_configuration( + self, + app_id: str, + version: int, + configuration: Optional[dict] = None, + ) -> dict: + """ + Update the outbound SCIM configuration attached to the given federated SSO app. + The version must match the currently stored version — otherwise the update is + rejected as a conflict. + + Args: + app_id (str): The federated SSO application id. + version (int): The currently stored version, used for optimistic concurrency. + configuration (dict): Optional updated provider-specific SCIM configuration. + + Return value (dict): + Return dict in the format + {"configuration": {"appId": , ...}} + + Raise: + AuthException: raised if update operation fails + """ + response = await self._http.post( + MgmtV1.outbound_scim_update_path, + body=OutboundSCIMBase._compose_update_body(app_id, version, configuration), + ) + return response.json() + + async def delete_configuration( + self, + app_id: str, + ): + """ + Delete the outbound SCIM configuration attached to the given federated SSO app. + IMPORTANT: This action is irreversible. Use carefully. + + Args: + app_id (str): The federated SSO application id. + + Raise: + AuthException: raised if deletion operation fails + """ + await self._http.post(MgmtV1.outbound_scim_delete_path, body={"appId": app_id}) + + async def load_configuration( + self, + app_id: str, + ) -> dict: + """ + Load the outbound SCIM configuration attached to the given federated SSO app. + + Args: + app_id (str): The federated SSO application id. + + Return value (dict): + Return dict in the format + {"configuration": {"appId": , "configuration": {...}, + "enabled": , "lastExportTime": , + "lastProcessingTime": , "failures": , + "version": }} + + Raise: + AuthException: raised if load operation fails + """ + response = await self._http.get(f"{MgmtV1.outbound_scim_load_path}/{app_id}") + return response.json() + + async def set_enabled( + self, + app_id: str, + enabled: bool, + ) -> dict: + """ + Enable or disable the outbound SCIM configuration attached to the given + federated SSO app. + + Args: + app_id (str): The federated SSO application id. + enabled (bool): Whether the SCIM configuration should be enabled. + + Return value (dict): + Return dict in the format + {"configuration": {"appId": , "enabled": , ...}} + + Raise: + AuthException: raised if the operation fails + """ + response = await self._http.post( + MgmtV1.outbound_scim_set_enabled_path, + body={"appId": app_id, "enabled": enabled}, + ) + return response.json() diff --git a/descope/mgmt.py b/descope/mgmt.py index cab52305f..97ed4375b 100644 --- a/descope/mgmt.py +++ b/descope/mgmt.py @@ -20,6 +20,7 @@ OutboundApplication, OutboundApplicationByToken, ) +from descope.management.outbound_scim import OutboundSCIM from descope.management.password import Password from descope.management.permission import Permission from descope.management.project import Project @@ -59,6 +60,7 @@ def __init__(self, http_client: HTTPClient, auth: Auth, fga_cache_url: Optional[ self._management_key = ManagementKey(http_client) self._outbound_application = OutboundApplication(http_client) self._outbound_application_by_token = OutboundApplicationByToken(http_client) + self._outbound_scim = OutboundSCIM(http_client) self._password = Password(http_client) self._permission = Permission(http_client) self._project = Project(http_client) @@ -163,6 +165,11 @@ def outbound_application_by_token(self): # No management key check for outbound_app_token (as authentication for those methods is done by inbound app token) return self._outbound_application_by_token + @property + def outbound_scim(self): + self._ensure_management_key("outbound_scim") + return self._outbound_scim + @property def descoper(self): self._ensure_management_key("descoper") diff --git a/descope/mgmt_async.py b/descope/mgmt_async.py index 790bb7d3d..78a896f4f 100644 --- a/descope/mgmt_async.py +++ b/descope/mgmt_async.py @@ -20,6 +20,7 @@ OutboundApplicationAsync, OutboundApplicationByTokenAsync, ) +from descope.management.outbound_scim_async import OutboundSCIMAsync from descope.management.password_async import PasswordAsync from descope.management.permission_async import PermissionAsync from descope.management.project_async import ProjectAsync @@ -61,6 +62,7 @@ def __init__(self, http_client: HTTPClientAsync, auth: AuthAsync, fga_cache_url: self._management_key = ManagementKeyAsync(http_client) self._outbound_application = OutboundApplicationAsync(http_client) self._outbound_application_by_token = OutboundApplicationByTokenAsync(http_client) + self._outbound_scim = OutboundSCIMAsync(http_client) self._password = PasswordAsync(http_client) self._permission = PermissionAsync(http_client) self._project = ProjectAsync(http_client) @@ -165,6 +167,11 @@ def outbound_application_by_token(self) -> OutboundApplicationByTokenAsync: # No management key check for outbound_app_token (as authentication for those methods is done by inbound app token) return self._outbound_application_by_token + @property + def outbound_scim(self) -> OutboundSCIMAsync: + self._ensure_management_key("outbound_scim") + return self._outbound_scim + @property def descoper(self) -> DescoperAsync: self._ensure_management_key("descoper") diff --git a/tests/management/test_outbound_scim.py b/tests/management/test_outbound_scim.py new file mode 100644 index 000000000..d914d6173 --- /dev/null +++ b/tests/management/test_outbound_scim.py @@ -0,0 +1,232 @@ +import pytest + +from descope import AuthException +from descope.management.common import MgmtV1 +from descope.management.outbound_scim import OutboundSCIM +from tests.common import DEFAULT_BASE_URL, default_headers +from tests.conftest import PROJECT_ID, assert_http_called, make_response +from tests.testutils import PUBLIC_KEY_DICT + +CONFIG_RESPONSE = { + "configuration": { + "appId": "app1", + "configuration": {"baseUrl": "https://scim.example.com"}, + "enabled": True, + "lastExportTime": 1719571200, + "lastProcessingTime": 1719571300, + "failures": 0, + "version": 3, + } +} + +MGMT_HEADERS = { + **default_headers, + "Authorization": f"Bearer {PROJECT_ID}:key", + "x-descope-project-id": PROJECT_ID, +} + + +class TestOutboundSCIM: + async def test_create_configuration_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + response = await client.invoke( + client.mgmt.outbound_scim.create_configuration( + "app1", + {"baseUrl": "https://scim.example.com"}, + ) + ) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_create_path}", + headers=MGMT_HEADERS, + params=None, + json={ + "appId": "app1", + "configuration": {"baseUrl": "https://scim.example.com"}, + }, + follow_redirects=False, + ) + + async def test_create_configuration_minimal_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + response = await client.invoke(client.mgmt.outbound_scim.create_configuration("app1")) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_create_path}", + headers=MGMT_HEADERS, + params=None, + json={"appId": "app1"}, + follow_redirects=False, + ) + + async def test_create_configuration_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.create_configuration("app1")) + + async def test_update_configuration_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + response = await client.invoke( + client.mgmt.outbound_scim.update_configuration( + "app1", + 3, + {"baseUrl": "https://scim.example.com"}, + ) + ) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_update_path}", + headers=MGMT_HEADERS, + params=None, + json={ + "appId": "app1", + "version": 3, + "configuration": {"baseUrl": "https://scim.example.com"}, + }, + follow_redirects=False, + ) + + async def test_update_configuration_no_config_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + await client.invoke(client.mgmt.outbound_scim.update_configuration("app1", 3)) + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_update_path}", + headers=MGMT_HEADERS, + params=None, + json={"appId": "app1", "version": 3}, + follow_redirects=False, + ) + + async def test_update_configuration_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.update_configuration("app1", 3, {})) + + async def test_delete_configuration_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response({})) as mock_post: + await client.invoke(client.mgmt.outbound_scim.delete_configuration("app1")) + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_delete_path}", + headers=MGMT_HEADERS, + params=None, + json={"appId": "app1"}, + follow_redirects=False, + ) + + async def test_delete_configuration_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.delete_configuration("app1")) + + async def test_load_configuration_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_get(make_response(CONFIG_RESPONSE)) as mock_get: + response = await client.invoke(client.mgmt.outbound_scim.load_configuration("app1")) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_get, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_load_path}/app1", + headers=MGMT_HEADERS, + params=None, + follow_redirects=True, + ) + + async def test_load_configuration_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_get(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.load_configuration("app1")) + + async def test_set_enabled_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + response = await client.invoke(client.mgmt.outbound_scim.set_enabled("app1", True)) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_set_enabled_path}", + headers=MGMT_HEADERS, + params=None, + json={"appId": "app1", "enabled": True}, + follow_redirects=False, + ) + + async def test_set_enabled_false_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + await client.invoke(client.mgmt.outbound_scim.set_enabled("app1", False)) + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_set_enabled_path}", + headers=MGMT_HEADERS, + params=None, + json={"appId": "app1", "enabled": False}, + follow_redirects=False, + ) + + async def test_set_enabled_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.set_enabled("app1", True)) + + def test_compose_create_body(self): + body = OutboundSCIM._compose_create_body("app1", {"baseUrl": "https://scim.example.com"}) + assert body == { + "appId": "app1", + "configuration": {"baseUrl": "https://scim.example.com"}, + } + + def test_compose_create_body_without_configuration(self): + body = OutboundSCIM._compose_create_body("app1") + assert body == {"appId": "app1"} + + def test_compose_update_body(self): + body = OutboundSCIM._compose_update_body( + "app1", + 5, + {"baseUrl": "https://scim.example.com"}, + ) + assert body == { + "appId": "app1", + "version": 5, + "configuration": {"baseUrl": "https://scim.example.com"}, + } + + def test_compose_update_body_without_configuration(self): + body = OutboundSCIM._compose_update_body("app1", 5) + assert body == {"appId": "app1", "version": 5}