groups.py

# The Okta software accompanied by this notice is provided pursuant to the following terms:
# Copyright © 2025-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.

import json
from typing import Optional
from urllib.parse import urlencode

from loguru import logger
from mcp.server.fastmcp import Context

from okta_mcp_server.server import mcp
from okta_mcp_server.tools.applications.applications import _camel_case_param, _safe_parse_app
from okta_mcp_server.utils.client import get_okta_client
from okta_mcp_server.utils.elicitation import DeleteConfirmation, elicit_or_fallback
from okta_mcp_server.utils.messages import DELETE_GROUP
from okta_mcp_server.utils.pagination import build_query_params, create_paginated_response, extract_after_cursor, paginate_all_results
from okta_mcp_server.utils.scope_guard import require_scopes
from okta_mcp_server.utils.validation import validate_ids


@mcp.tool()
@require_scopes("okta.groups.read", error_return_type="list")
async def list_groups(
    ctx: Context,
    search: str = "",
    filter: Optional[str] = None,
    q: Optional[str] = None,
    fetch_all: bool = False,
    after: Optional[str] = None,
    limit: Optional[int] = None,
) -> dict:
    """List all the groups from the Okta organization with pagination support.
    If search, filter, or q is specified, it will list only those groups that satisfy the condition.

    Parameters:
        search (str, optional): The value of the search string when searching for some specific set of groups.
        filter (str, optional): A filter string to filter groups by Okta profile attributes.
        q (str, optional): A query string to search groups by Okta profile attributes.
        fetch_all (bool, optional): If True, automatically fetch all pages of results. Default: False.
        after (str, optional): Pagination cursor for fetching results after this point.
        limit (int, optional): Maximum number of groups to return per page (min 20, max 100).
        The search, filter, and q are performed on group profile attributes.

    Examples:
        For pagination:
        - First call: list_groups(search="profile.name sw \"Engineering\"")
        - Next page: list_groups(search="profile.name sw \"Engineering\"", after="cursor_value")
        - All pages: list_groups(search="profile.name sw \"Engineering\"", fetch_all=True)

    Returns:
        Dict containing:
        - items: List of group objects
        - total_fetched: Number of groups returned
        - has_more: Boolean indicating if more results are available
        - next_cursor: Cursor for the next page (if has_more is True)
        - fetch_all_used: Boolean indicating if fetch_all was used
        - pagination_info: Additional pagination metadata (when fetch_all=True)
    """
    logger.info("Listing groups from Okta organization")
    logger.debug(
        f"Search: '{search}', Filter: '{filter}', Q: '{q}', fetch_all: {fetch_all}, after: '{after}', limit: {limit}"
    )

    # Enforce a consistent default page size when no limit is provided.
    if limit is None:
        limit = 20

    # Validate limit parameter range
    if limit < 20:
        logger.warning(f"Limit {limit} is below minimum (20), setting to 20")
        limit = 20
    elif limit > 100:
        logger.warning(f"Limit {limit} exceeds maximum (100), setting to 100")
        limit = 100

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)
        query_params = build_query_params(search=search, filter=filter, q=q, after=after, limit=limit)

        logger.debug("Calling Okta API to list groups")
        groups, response, err = await client.list_groups(**query_params)

        if err:
            logger.error(f"Okta API error while listing groups: {err}")
            return {"error": f"Error: {err}"}

        if not groups:
            logger.info("No groups found")
            return create_paginated_response([], response, fetch_all)

        _has_more = (hasattr(response, "has_next") and response.has_next()) or bool(extract_after_cursor(response))
        if fetch_all and response and _has_more:
            logger.info(f"fetch_all=True, auto-paginating from initial {len(groups)} groups")

            async def _next_page(cursor):
                p = dict(query_params)
                p["after"] = cursor
                return await client.list_groups(**p)

            async def _on_page(pages, total):
                await ctx.info(f"Fetching groups... {total} fetched so far ({pages} pages)")

            all_groups, pagination_info = await paginate_all_results(
                response, groups, next_page_fn=_next_page, on_page=_on_page
            )

            logger.info(
                f"Successfully retrieved {len(all_groups)} groups across {pagination_info['pages_fetched']} pages"
            )
            return create_paginated_response(
                all_groups, response, fetch_all_used=True, pagination_info=pagination_info
            )
        else:
            logger.info(f"Successfully retrieved {len(groups)} groups")
            return create_paginated_response(groups, response, fetch_all_used=fetch_all)

    except Exception as e:
        logger.error(f"Exception while listing groups: {type(e).__name__}: {e}")
        return {"error": f"Exception: {e}"}


@mcp.tool()
@require_scopes("okta.groups.read", error_return_type="list")
@validate_ids("group_id")
async def get_group(group_id: str, ctx: Context = None) -> list:
    """Get a group by ID from the Okta organization

    This tool retrieves a group by its ID from the Okta organization.

    Parameters:
        group_id (str, required): The ID of the group to retrieve.

    Returns:
        List containing the group details.
    """
    logger.info(f"Getting group with ID: {group_id}")

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)
        logger.debug(f"Calling Okta API to get group {group_id}")

        group, _, err = await client.get_group(group_id)

        if err:
            logger.error(f"Okta API error while getting group {group_id}: {err}")
            return [f"Error: {err}"]

        logger.info(f"Successfully retrieved group: {group_id}")
        return [group]
    except Exception as e:
        logger.error(f"Exception while getting group {group_id}: {type(e).__name__}: {e}")
        return [f"Exception: {e}"]


@mcp.tool()
@require_scopes("okta.groups.manage", error_return_type="list")
async def create_group(profile: dict, ctx: Context = None) -> list:
    """Create a group in the Okta organization.

    This tool creates a new group in the Okta organization with the provided profile.

    Parameters:
        profile (dict, required): The profile of the group to create.

    Returns:
        List containing the created group details.
    """
    logger.info("Creating new group in Okta organization")
    logger.debug(f"Group profile: name={profile.get('name', 'N/A')}, description={profile.get('description', 'N/A')}")

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)
        # Wrap the profile in a dict with 'profile' key as required by Okta SDK
        logger.debug("Calling Okta API to create group")

        group, _, err = await client.add_group({"profile": profile})

        if err:
            logger.error(f"Okta API error while creating group: {err}")
            return {"error": f"Error: {err}"}

        profile_instance = getattr(group.profile, "actual_instance", None) if hasattr(group, "profile") else None
        group_name = getattr(profile_instance, "name", "N/A") if profile_instance is not None else "N/A"
        logger.info(f"Successfully created group: {group.id} ({group_name})")
        return [group]
    except Exception as e:
        logger.error(f"Exception while creating group: {type(e).__name__}: {e}")
        return [f"Exception: {e}"]


@mcp.tool()
@require_scopes("okta.groups.manage", error_return_type="list")
@validate_ids("group_id")
async def delete_group(group_id: str, ctx: Context = None) -> list:
    """Delete a group by ID from the Okta organization.

    This tool deletes a group by its ID from the Okta organization.
    The user will be asked for confirmation before the deletion proceeds.

    Parameters:
        group_id (str, required): The ID of the group to delete.

    Returns:
        List containing the result of the deletion operation.
    """
    logger.warning(f"Deletion requested for group {group_id}")

    fallback_payload = {
        "confirmation_required": True,
        "message": (
            f"To confirm deletion of group {group_id}, please call the "
            f"'confirm_delete_group' tool with group_id='{group_id}' and "
            f"confirmation='DELETE'."
        ),
        "group_id": group_id,
        "tool_to_use": "confirm_delete_group",
    }

    outcome = await elicit_or_fallback(
        ctx,
        message=DELETE_GROUP.format(group_id=group_id),
        schema=DeleteConfirmation,
        fallback_payload=fallback_payload,
    )

    if not outcome.used_elicitation:
        logger.info(f"Elicitation unavailable for group {group_id} — returning fallback confirmation prompt")
        return [outcome.fallback_response]

    if not outcome.confirmed:
        logger.info(f"Group deletion cancelled for {group_id}")
        return [{"message": "Group deletion cancelled by user."}]

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)
        logger.debug(f"Calling Okta API to delete group {group_id}")

        result = await client.delete_group(group_id)
        err = result[-1]

        if err:
            logger.error(f"Okta API error while deleting group {group_id}: {err}")
            return [{"error": f"Error: {err}"}]

        logger.info(f"Successfully deleted group: {group_id}")
        return [{"message": f"Group {group_id} deleted successfully"}]
    except Exception as e:
        logger.error(f"Exception while deleting group {group_id}: {type(e).__name__}: {e}")
        return [{"error": f"Exception: {e}"}]


@mcp.tool()
@require_scopes("okta.groups.manage", error_return_type="list")
@validate_ids("group_id")
async def confirm_delete_group(group_id: str, confirmation: str, ctx: Context = None) -> list:
    """Confirm and execute group deletion after receiving confirmation.

    .. deprecated::
        This tool exists for backward compatibility with clients that do not
        support MCP elicitation.  New clients should rely on the built-in
        elicitation prompt in ``delete_group`` instead.

    This function MUST ONLY be called after the human user has explicitly typed 'DELETE' as confirmation.
    NEVER call this function automatically after delete_group.

    Parameters:
        group_id (str, required): The ID of the group to delete.
        confirmation (str, required): Must be 'DELETE' to confirm deletion.

    Returns:
        List containing the result of the deletion operation.
    """
    logger.info(f"Processing deletion confirmation for group {group_id} (deprecated flow)")

    if confirmation != "DELETE":
        logger.warning(f"Group deletion cancelled for {group_id} - incorrect confirmation")
        return [{"error": "Deletion cancelled. Confirmation 'DELETE' was not provided correctly."}]

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)
        logger.debug(f"Calling Okta API to delete group {group_id}")

        result = await client.delete_group(group_id)
        err = result[-1]

        if err:
            logger.error(f"Okta API error while deleting group {group_id}: {err}")
            return [{"error": str(err)}]

        logger.info(f"Successfully deleted group: {group_id}")
        return [{"message": f"Group {group_id} deleted successfully"}]
    except Exception as e:
        logger.error(f"Exception while deleting group {group_id}: {type(e).__name__}: {e}")
        return [{"error": str(e)}]


@mcp.tool()
@require_scopes("okta.groups.manage", error_return_type="list")
@validate_ids("group_id")
async def update_group(group_id: str, profile: dict, ctx: Context = None) -> list:
    """Update a group by ID in the Okta organization.

    This tool updates a group by its ID with the provided profile.

    Parameters:
        group_id (str, required): The ID of the group to update.
        profile (dict, required): The new profile for the group.

    Returns:
        List containing the updated group details.
    """
    logger.info(f"Updating group with ID: {group_id}")
    logger.debug(f"Updated fields: {list(profile.keys())}")

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)
        # Wrap the profile in a dict with 'profile' key as required by Okta SDK
        logger.debug(f"Calling Okta API to update group {group_id}")

        group, _, err = await client.replace_group(group_id, {"profile": profile})

        if err:
            logger.error(f"Okta API error while updating group {group_id}: {err}")
            return [f"Error: {err}"]

        logger.info(f"Successfully updated group: {group_id}")
        return [group]
    except Exception as e:
        logger.error(f"Exception while updating group {group_id}: {type(e).__name__}: {e}")
        return [f"Exception: {e}"]


@mcp.tool()
@require_scopes("okta.groups.read", error_return_type="list")
@validate_ids("group_id", error_return_type="dict")
async def list_group_users(
    group_id: str,
    ctx: Context = None,
    fetch_all: bool = False,
    after: Optional[str] = None,
    limit: Optional[int] = None,
) -> dict:
    """List all users in a group by ID from the Okta organization with pagination support.

    This tool retrieves all users in a group by its ID from the Okta organization.

    Parameters:
        group_id (str, required): The ID of the group to retrieve users from.
        fetch_all (bool, optional): If True, automatically fetch all pages of results. Default: False.
        after (str, optional): Pagination cursor for fetching results after this point.
        limit (int, optional): Maximum number of users to return per page (min 20, max 100).

    Examples:
        For pagination:
        - First call: list_group_users("group_id")
        - Next page: list_group_users("group_id", after="cursor_value")
        - All pages: list_group_users("group_id", fetch_all=True)

    Returns:
        Dict containing:
        - items: List of user objects in the group
        - total_fetched: Number of users returned
        - has_more: Boolean indicating if more results are available
        - next_cursor: Cursor for the next page (if has_more is True)
        - fetch_all_used: Boolean indicating if fetch_all was used
        - pagination_info: Additional pagination metadata (when fetch_all=True)
    """
    logger.info(f"Listing users in group: {group_id}")
    logger.debug(f"fetch_all: {fetch_all}, after: '{after}', limit: {limit}")

    # Enforce a consistent default page size when no limit is provided.
    if limit is None:
        limit = 20

    # Validate limit parameter range
    if limit < 20:
        logger.warning(f"Limit {limit} is below minimum (20), setting to 20")
        limit = 20
    elif limit > 100:
        logger.warning(f"Limit {limit} exceeds maximum (100), setting to 100")
        limit = 100

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)
        logger.debug(f"Calling Okta API to list users in group {group_id}")

        query_params = build_query_params(after=after, limit=limit)
        users, response, err = await client.list_group_users(group_id, **query_params)

        if err:
            logger.error(f"Okta API error while listing group users for {group_id}: {err}")
            return {"error": f"Error: {err}"}

        if not users:
            logger.info(f"No users found in group {group_id}")
            return create_paginated_response([], response, fetch_all)

        _has_more = (hasattr(response, "has_next") and response.has_next()) or bool(extract_after_cursor(response))
        if fetch_all and response and _has_more:
            logger.info(f"fetch_all=True, auto-paginating from initial {len(users)} users in group {group_id}")

            async def _next_page(cursor):
                p = dict(query_params)
                p["after"] = cursor
                return await client.list_group_users(group_id, **p)

            async def _on_page(pages, total):
                await ctx.info(f"Fetching group users... {total} fetched so far ({pages} pages)")

            all_users, pagination_info = await paginate_all_results(
                response, users, next_page_fn=_next_page, on_page=_on_page
            )

            pages_fetched = pagination_info["pages_fetched"]
            logger.info(
                f"Successfully retrieved {len(all_users)} users from group {group_id} across {pages_fetched} pages"
            )
            return create_paginated_response(all_users, response, fetch_all_used=True, pagination_info=pagination_info)
        else:
            logger.info(f"Successfully retrieved {len(users)} users from group {group_id}")
            return create_paginated_response(users, response, fetch_all_used=fetch_all)

    except Exception as e:
        logger.error(f"Exception while listing users in group {group_id}: {type(e).__name__}: {e}")
        return {"error": f"Exception: {e}"}


@mcp.tool()
@require_scopes("okta.groups.read", error_return_type="dict")
@validate_ids("group_id", error_return_type="dict")
async def list_group_apps(
    group_id: str,
    ctx: Context = None,
    after: Optional[str] = None,
    limit: Optional[int] = None,
    fetch_all: bool = False,
) -> dict:
    """List all applications assigned to a group with pagination support.

    Parameters:
        group_id (str, required): The ID of the group to retrieve applications from.
        after (str, optional): Pagination cursor for the next page.
        limit (int, optional): Maximum number of apps to return per page (max 200). Default: 20.
        fetch_all (bool, optional): If True, automatically fetch all pages. Default: False.

    Returns:
        Dict containing:
            - items (List): List of application objects
            - total_fetched (int): Number of apps returned
            - has_more (bool): Whether more results are available
            - next_cursor (str | None): Cursor for the next page
            - fetch_all_used (bool): Whether fetch_all was used
            - pagination_info (Dict): Detailed pagination metadata (when fetch_all=True)
    """
    logger.info(f"Listing applications assigned to group: {group_id}")
    logger.debug(f"fetch_all: {fetch_all}, after: '{after}', limit: {limit}")

    if limit is None:
        limit = 20
    if limit > 200:
        logger.warning(f"Limit {limit} exceeds maximum (200), setting to 200")
        limit = 200

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)
        effective_limit = 200 if fetch_all else limit
        query_params = build_query_params(after=after, limit=effective_limit)
        logger.debug(f"Calling Okta API to list applications for group {group_id}")

        async def _fetch_group_apps_page(params):
            """Fetch one page of /api/v1/groups/{id}/apps and parse each app permissively.

            The typed ``client.list_assigned_applications_for_group`` validates the whole
            page into strict SDK models in one pass, so a single non-conforming app — a
            SAML app with a partial ``settings.signOn`` or a custom SWA whose ``name`` is
            outside the SDK enum — aborts the entire response. Fetching the raw page
            through the request executor and parsing per item via ``_safe_parse_app``
            avoids that. Mirrors the resilient path in ``list_applications``.
            """
            executor = client.get_request_executor()
            query_string = urlencode(
                {
                    _camel_case_param(k): ("true" if v is True else "false" if v is False else v)
                    for k, v in params.items()
                }
            )
            url = f"/api/v1/groups/{group_id}/apps" + (f"?{query_string}" if query_string else "")
            request, request_err = await executor.create_request(
                method="GET", url=url, body={}, headers={}, oauth=False
            )
            if request_err:
                return None, None, request_err

            page_response, response_body, response_err = await executor.execute(request)
            if response_err:
                return None, page_response, response_err

            raw_items = json.loads(response_body) if response_body else []
            parsed_items = [_safe_parse_app(item) for item in raw_items]
            return parsed_items, page_response, None

        apps, response, err = await _fetch_group_apps_page(query_params)

        if err:
            logger.error(f"Okta API error while listing applications for group {group_id}: {err}")
            return {"error": str(err)}

        if not apps:
            logger.info(f"No applications found for group {group_id}")
            return create_paginated_response([], response, fetch_all_used=fetch_all)

        _has_more = (hasattr(response, "has_next") and response.has_next()) or bool(extract_after_cursor(response))
        if fetch_all and response and _has_more:
            logger.info(f"fetch_all=True, auto-paginating from initial {len(apps)} apps for group {group_id}")

            async def _next_page(cursor):
                p = {k: v for k, v in query_params.items() if k != "after"}
                p["after"] = cursor
                return await _fetch_group_apps_page(p)

            async def _on_page(pages, total):
                logger.info(f"[list_group_apps] Page {pages} fetched — {total} apps so far")

            all_apps, pagination_info = await paginate_all_results(
                response, apps, next_page_fn=_next_page, on_page=_on_page
            )
            logger.info(f"Successfully retrieved {len(all_apps)} apps for group {group_id} across {pagination_info['pages_fetched']} pages")
            return create_paginated_response(all_apps, response, fetch_all_used=True, pagination_info=pagination_info)

        logger.info(f"Successfully retrieved {len(apps)} applications for group {group_id}")
        return create_paginated_response(apps, response, fetch_all_used=fetch_all)
    except Exception as e:
        logger.error(f"Exception while listing applications for group {group_id}: {type(e).__name__}: {e}")
        return {"error": str(e)}


@mcp.tool()
@require_scopes("okta.groups.manage", error_return_type="list")
@validate_ids("group_id", "user_id")
async def add_user_to_group(group_id: str, user_id: str, ctx: Context = None) -> list:
    """Add a user to a group by ID in the Okta organization.

    This tool adds a user to a group by its ID in the Okta organization.

    Parameters:
        group_id (str, required): The ID of the group to add the user to.
        user_id (str, required): The ID of the user to add to the group.

    Returns:
        List containing the result of the addition operation.
    """
    logger.info(f"Adding user {user_id} to group {group_id}")

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)

        # Idempotency check: use list_user_groups(user_id) and check if group_id is
        # present. This is always a single API call regardless of group size because
        # users typically belong to O(10-50) groups, whereas groups can have thousands
        # of members. Querying from the user side is orders of magnitude cheaper than
        # paginating all members of the group via list_group_users.
        # SDK: list_user_groups(id) accepts only the user id — no pagination params —
        # and returns the full list in one response.
        logger.debug(f"Checking if user {user_id} is already a member of group {group_id}")
        user_groups, _, groups_err = await client.list_user_groups(user_id)
        if not groups_err and user_groups:
            if any(g.id == group_id for g in user_groups):
                logger.info(f"User {user_id} is already a member of group {group_id}")
                return [f"User {user_id} is already a member of group {group_id}"]

        logger.debug(f"Calling Okta API to add user {user_id} to group {group_id}")
        result = await client.assign_user_to_group(group_id, user_id)
        err = result[-1]

        if err:
            logger.error(f"Okta API error while adding user {user_id} to group {group_id}: {err}")
            return [f"Error: {err}"]

        logger.info(f"Successfully added user {user_id} to group {group_id}")
        return [f"User {user_id} added to group {group_id} successfully"]
    except Exception as e:
        logger.error(f"Exception while adding user {user_id} to group {group_id}: {type(e).__name__}: {e}")
        return [f"Exception: {e}"]


@mcp.tool()
@require_scopes("okta.groups.manage", error_return_type="list")
@validate_ids("group_id", "user_id")
async def remove_user_from_group(group_id: str, user_id: str, ctx: Context = None) -> list:
    """Remove a user from a group by ID in the Okta organization.

    This tool removes a user from a group by its ID in the Okta organization.

    Parameters:
        group_id (str, required): The ID of the group to remove the user from.
        user_id (str, required): The ID of the user to remove from the group.

    Returns:
        List containing the result of the removal operation.
    """
    logger.info(f"Removing user {user_id} from group {group_id}")

    manager = ctx.request_context.lifespan_context.okta_auth_manager

    try:
        client = await get_okta_client(manager)
        logger.debug(f"Calling Okta API to remove user {user_id} from group {group_id}")

        result = await client.unassign_user_from_group(group_id, user_id)
        err = result[-1]

        if err:
            logger.error(f"Okta API error while removing user {user_id} from group {group_id}: {err}")
            return [f"Error: {err}"]

        logger.info(f"Successfully removed user {user_id} from group {group_id}")
        return [f"User {user_id} removed from group {group_id} successfully"]
    except Exception as e:
        logger.error(f"Exception while removing user {user_id} from group {group_id}: {type(e).__name__}: {e}")
        return [f"Exception: {e}"]