diff --git a/README.md b/README.md index 04351f0..a008538 100644 --- a/README.md +++ b/README.md @@ -487,6 +487,9 @@ The Okta MCP Server provides the following tools for LLMs to interact with your | `delete_application` | Delete an application (prompts for confirmation) | - `Delete the old legacy application`
- `Remove the unused test application`
- `Clean up deprecated integrations` | | `activate_application` | Activate an application | - `Activate the new HR application`
- `Enable the Salesforce integration`
- `Turn on the mobile app for users` | | `deactivate_application` | Deactivate an application (prompts for confirmation) | - `Deactivate the legacy CRM application`
- `Temporarily disable the mobile app`
- `Turn off access to the test environment` | +| `list_catalog_apps` | Browse the OIN app catalog (find an app's `name` + features) | - `Find a SCIM 2.0 test app in the catalog`
- `Search the OIN catalog for Slack` | +| `get_catalog_app` | Get a single OIN catalog app definition (incl. schema) | - `Show the catalog definition for the SCIM 2.0 test app` | +| `install_oin_app` | Install an instance of an OIN catalog app (e.g. a provisioning-capable SCIM app) | - `Install the SCIM 2.0 test app as "HR Directory Sync"`
- `Add the Slack OIN app to my org` | ### Policies @@ -658,8 +661,8 @@ The Okta MCP Server uses a **scope-based tool loading** mechanism to ensure that | `okta.users.manage` | `create_user`, `update_user`, `deactivate_user`, `delete_deactivated_user` | | `okta.groups.read` | `list_groups`, `get_group`, `list_group_users`, `list_group_apps` | | `okta.groups.manage` | `create_group`, `update_group`, `delete_group`, `add_user_to_group`, `remove_user_from_group` | -| `okta.apps.read` | `list_applications`, `get_application` | -| `okta.apps.manage` | `create_application`, `update_application`, `delete_application`, `activate_application`, `deactivate_application` | +| `okta.apps.read` | `list_applications`, `get_application`, `list_catalog_apps`, `get_catalog_app` | +| `okta.apps.manage` | `create_application`, `update_application`, `delete_application`, `activate_application`, `deactivate_application`, `install_oin_app` | | `okta.policies.read` | `list_policies`, `get_policy`, `list_policy_rules`, `get_policy_rule` | | `okta.policies.manage` | `create_policy`, `update_policy`, `delete_policy`, `activate_policy`, `deactivate_policy`, `create_policy_rule`, `update_policy_rule`, `delete_policy_rule`, `activate_policy_rule`, `deactivate_policy_rule` | | `okta.deviceAssurance.read` | `list_device_assurance_policies`, `get_device_assurance_policy` | diff --git a/src/okta_mcp_server/tools/applications/applications.py b/src/okta_mcp_server/tools/applications/applications.py index 70c3e35..c94833b 100644 --- a/src/okta_mcp_server/tools/applications/applications.py +++ b/src/okta_mcp_server/tools/applications/applications.py @@ -5,7 +5,9 @@ # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and limitations under the License. +import json from typing import Any, Dict, Optional +from urllib.parse import urlencode import okta.models as okta_models from loguru import logger @@ -444,3 +446,159 @@ async def deactivate_application(ctx: Context, app_id: str) -> list: except Exception as e: logger.error(f"Exception while deactivating application {app_id}: {type(e).__name__}: {e}") return [f"Exception: {e}"] + + +# --------------------------------------------------------------------------- +# OIN catalog & app installation +# --------------------------------------------------------------------------- + + +@mcp.tool() +@require_scopes("okta.apps.read") +async def list_catalog_apps(ctx: Context, q: Optional[str] = None, limit: Optional[int] = None) -> Any: + """Browse the Okta Integration Network (OIN) app catalog. + + Use this to discover an app's ``name`` (the catalog key) to pass to + install_oin_app. Apps that support outbound provisioning list a provisioning + capability in their ``features`` (e.g. a SCIM 2.0 test app). A plain custom + SAML/OIDC app created with create_application cannot do provisioning; you need + an installed instance of a provisioning-capable catalog app. + + Parameters: + q (str, optional): Filters the catalog by app name/keyword (e.g. "scim") + limit (int, optional): Maximum number of catalog entries to return + + Returns: + Dict with a ``catalog_apps`` list (each has ``name``, ``displayName``, + ``features``, ``signOnModes``), or error information. + """ + logger.info(f"Browsing OIN catalog (q='{q}', limit={limit})") + + manager = ctx.request_context.lifespan_context.okta_auth_manager + + try: + client = await get_okta_client(manager) + params: Dict[str, Any] = {} + if q: + params["q"] = q + if limit: + params["limit"] = limit + query_string = urlencode(params) + url = "/api/v1/catalog/apps" + (f"?{query_string}" if query_string else "") + + executor = client.get_request_executor() + request, err = await executor.create_request(method="GET", url=url, body={}, headers={}, oauth=False) + if err: + logger.error(f"Error building catalog request: {err}") + return {"error": str(err)} + + _, response_body, err = await executor.execute(request) + if err: + logger.error(f"Okta API error while browsing catalog: {err}") + return {"error": str(err)} + + logger.info("Successfully retrieved OIN catalog page") + return {"catalog_apps": json.loads(response_body) if response_body else []} + except Exception as e: + logger.error(f"Exception while browsing OIN catalog: {type(e).__name__}: {e}") + return {"error": str(e)} + + +@mcp.tool() +@require_scopes("okta.apps.read") +@validate_ids("app_name", error_return_type="dict") +async def get_catalog_app(ctx: Context, app_name: str) -> Any: + """Get a single OIN catalog app definition (including its provisioning schema). + + Parameters: + app_name (str, required): The catalog app key (from list_catalog_apps) + + Returns: + Dict with the catalog app definition, or error information. + """ + logger.info(f"Getting OIN catalog app: {app_name}") + + manager = ctx.request_context.lifespan_context.okta_auth_manager + + try: + client = await get_okta_client(manager) + url = f"/api/v1/catalog/apps/{app_name}?expand=schema" + + executor = client.get_request_executor() + request, err = await executor.create_request(method="GET", url=url, body={}, headers={}, oauth=False) + if err: + logger.error(f"Error building catalog-app request for {app_name}: {err}") + return {"error": str(err)} + + _, response_body, err = await executor.execute(request) + if err: + logger.error(f"Okta API error while getting catalog app {app_name}: {err}") + return {"error": str(err)} + + logger.info(f"Successfully retrieved OIN catalog app: {app_name}") + return json.loads(response_body) if response_body else {} + except Exception as e: + logger.error(f"Exception while getting catalog app {app_name}: {type(e).__name__}: {e}") + return {"error": str(e)} + + +@mcp.tool() +@require_scopes("okta.apps.manage") +async def install_oin_app( + ctx: Context, + name: str, + label: str, + sign_on_mode: str, + settings: Optional[Dict[str, Any]] = None, + activate: bool = True, +) -> Any: + """Install an instance of an OIN catalog app (e.g. a provisioning-capable SCIM app). + + Unlike create_application (which builds custom apps), this preserves the + catalog ``name`` key. The typed SDK create path strips ``name`` from the + request body, so a catalog/OIN app can't be installed through create_application; + this issues the request directly. Provisioning capability is determined by the + catalog app definition at install time — it cannot be added to a custom app + afterwards. Discover ``name`` and the allowed ``signOnMode`` values via + list_catalog_apps / get_catalog_app. + + Parameters: + name (str, required): The OIN catalog app key (e.g. from list_catalog_apps) + label (str, required): Display label for the installed instance + sign_on_mode (str, required): A sign-on mode the catalog app supports + (e.g. SAML_2_0, OPENID_CONNECT, SECURE_PASSWORD_STORE, BOOKMARK) + settings (dict, optional): Additional app settings/profile some OIN apps require + activate (bool, optional): Activate the app on creation. Defaults to True. + + Returns: + Dict with the installed application, or error information. + """ + logger.info(f"Installing OIN app '{name}' (label='{label}', signOnMode='{sign_on_mode}')") + + manager = ctx.request_context.lifespan_context.okta_auth_manager + + try: + client = await get_okta_client(manager) + app_body: Dict[str, Any] = {"name": name, "label": label, "signOnMode": sign_on_mode} + if settings: + app_body["settings"] = settings + + query_string = urlencode({"activate": "true" if activate else "false"}) + url = f"/api/v1/apps?{query_string}" + + executor = client.get_request_executor() + request, err = await executor.create_request(method="POST", url=url, body=app_body, headers={}, oauth=False) + if err: + logger.error(f"Error building OIN install request for '{name}': {err}") + return {"error": str(err)} + + _, response_body, err = await executor.execute(request) + if err: + logger.error(f"Okta API error while installing OIN app '{name}': {err}") + return {"error": str(err)} + + logger.info(f"Successfully installed OIN app '{name}'") + return json.loads(response_body) if response_body else {} + except Exception as e: + logger.error(f"Exception while installing OIN app '{name}': {type(e).__name__}: {e}") + return {"error": str(e)} diff --git a/src/okta_mcp_server/utils/scope_registry.py b/src/okta_mcp_server/utils/scope_registry.py index b713588..197eefb 100644 --- a/src/okta_mcp_server/utils/scope_registry.py +++ b/src/okta_mcp_server/utils/scope_registry.py @@ -59,6 +59,9 @@ "confirm_delete_application": "okta.apps.manage", "activate_application": "okta.apps.manage", "deactivate_application": "okta.apps.manage", + "list_catalog_apps": "okta.apps.read", + "get_catalog_app": "okta.apps.read", + "install_oin_app": "okta.apps.manage", # ------------------------------------------------------------------ # Policies (src/okta_mcp_server/tools/policies/policies.py) # ------------------------------------------------------------------ diff --git a/tests/test_oin_catalog.py b/tests/test_oin_catalog.py new file mode 100644 index 0000000..bfe2722 --- /dev/null +++ b/tests/test_oin_catalog.py @@ -0,0 +1,131 @@ +# The Okta software accompanied by this notice is provided pursuant to the following terms: +# Copyright © 2026-Present, Okta, Inc. +# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0. +# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and limitations under the License. + +"""Tests for OIN catalog browse + install tools.""" + +from __future__ import annotations + +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from okta_mcp_server.tools.applications.applications import ( + get_catalog_app, + install_oin_app, + list_catalog_apps, +) + + +def _make_ctx(): + from tests.conftest import FakeLifespanContext, FakeOktaAuthManager + + request_context = MagicMock() + request_context.lifespan_context = FakeLifespanContext( + okta_auth_manager=FakeOktaAuthManager() + ) + ctx = MagicMock() + ctx.request_context = request_context + return ctx + + +def _client_returning(body, execute_error=None): + executor = MagicMock() + executor.create_request = AsyncMock(return_value=({"method": "X"}, None)) + if execute_error is not None: + executor.execute = AsyncMock(return_value=(None, None, execute_error)) + else: + executor.execute = AsyncMock(return_value=(MagicMock(), body, None)) + client = MagicMock() + client.get_request_executor = MagicMock(return_value=executor) + return client + + +class TestListCatalogApps: + @pytest.mark.asyncio + @patch("okta_mcp_server.tools.applications.applications.get_okta_client") + async def test_returns_catalog_apps_and_passes_query(self, mock_get_client): + catalog = [{"name": "scim2testapp", "displayName": "SCIM 2.0 Test App", "features": ["IMPORT_NEW_USERS"]}] + client = _client_returning(json.dumps(catalog)) + mock_get_client.return_value = client + + result = await list_catalog_apps(ctx=_make_ctx(), q="scim") + + assert result == {"catalog_apps": catalog} + kwargs = client.get_request_executor.return_value.create_request.call_args.kwargs + assert kwargs["method"] == "GET" + assert "/api/v1/catalog/apps?q=scim" in kwargs["url"] + + @pytest.mark.asyncio + @patch("okta_mcp_server.tools.applications.applications.get_okta_client") + async def test_error_is_returned(self, mock_get_client): + mock_get_client.return_value = _client_returning(None, execute_error="403 forbidden") + + result = await list_catalog_apps(ctx=_make_ctx()) + + assert result == {"error": "403 forbidden"} + + +class TestGetCatalogApp: + @pytest.mark.asyncio + @patch("okta_mcp_server.tools.applications.applications.get_okta_client") + async def test_returns_app_definition_with_schema_expand(self, mock_get_client): + client = _client_returning(json.dumps({"name": "scim2testapp", "status": "ACTIVE"})) + mock_get_client.return_value = client + + result = await get_catalog_app(ctx=_make_ctx(), app_name="scim2testapp") + + assert result["name"] == "scim2testapp" + url = client.get_request_executor.return_value.create_request.call_args.kwargs["url"] + assert url.endswith("/api/v1/catalog/apps/scim2testapp?expand=schema") + + +class TestInstallOinApp: + @pytest.mark.asyncio + @patch("okta_mcp_server.tools.applications.applications.get_okta_client") + async def test_posts_body_with_name_preserved(self, mock_get_client): + created = {"id": "0oaNEW0000001", "name": "scim2testapp", "label": "HR Directory Sync", "status": "ACTIVE"} + client = _client_returning(json.dumps(created)) + mock_get_client.return_value = client + + result = await install_oin_app( + ctx=_make_ctx(), name="scim2testapp", label="HR Directory Sync", sign_on_mode="SAML_2_0" + ) + + assert result == created + kwargs = client.get_request_executor.return_value.create_request.call_args.kwargs + assert kwargs["method"] == "POST" + assert kwargs["url"].startswith("/api/v1/apps?") + assert "activate=true" in kwargs["url"] + # The catalog `name` MUST be in the body — the whole point (the typed SDK path drops it). + assert kwargs["body"] == {"name": "scim2testapp", "label": "HR Directory Sync", "signOnMode": "SAML_2_0"} + + @pytest.mark.asyncio + @patch("okta_mcp_server.tools.applications.applications.get_okta_client") + async def test_settings_and_activate_false_forwarded(self, mock_get_client): + client = _client_returning(json.dumps({})) + mock_get_client.return_value = client + + await install_oin_app( + ctx=_make_ctx(), name="scim2testapp", label="X", sign_on_mode="SAML_2_0", + settings={"app": {"acsUrl": "https://x"}}, activate=False, + ) + + kwargs = client.get_request_executor.return_value.create_request.call_args.kwargs + assert "activate=false" in kwargs["url"] + assert kwargs["body"]["settings"] == {"app": {"acsUrl": "https://x"}} + + @pytest.mark.asyncio + @patch("okta_mcp_server.tools.applications.applications.get_okta_client") + async def test_error_is_returned(self, mock_get_client): + mock_get_client.return_value = _client_returning(None, execute_error="400 invalid app name") + + result = await install_oin_app( + ctx=_make_ctx(), name="bogus", label="X", sign_on_mode="SAML_2_0" + ) + + assert result == {"error": "400 invalid app name"}