Microsoft Graph Groups

import demistomock as demisto
import urllib3
from CommonServerPython import *
from MicrosoftApiModule import *  # noqa: E402

# Disable insecure warnings

urllib3.disable_warnings()

INTEGRATION_CONTEXT_NAME = "MSGraphGroups"
NO_OUTPUTS: dict = {}
APP_NAME = "ms-graph-groups"


def camel_case_to_readable(text: str) -> str:
    """'camelCase' -> 'Camel Case'

    Args:
        text: the text to transform

    Returns:
        A Camel Cased string.
    """
    if text == "id":
        return "ID"
    return "".join(" " + char if char.isupper() else char.strip() for char in text).strip().title()


def parse_outputs(groups_data: dict[str, str]) -> tuple[dict, dict]:
    """Parse group data as received from Microsoft Graph API into Demisto's conventions

    Args:
        groups_data: a dictionary containing the group data

    Returns:
        A Camel Cased dictionary with the relevant fields.
        groups_readable: for the human readable
        groups_outputs: for the entry context
    """
    # Unnecessary fields, dropping as to not load the incident context.
    fields_to_drop = [
        "@odata.context",
        "@odata.nextLink",
        "@odata.deltaLink",
        "@odata.type",
        "@removed",
        "resourceProvisioningOptions",
        "securityIdentifier",
        "onPremisesSecurityIdentifier",
        "onPremisesNetBiosName",
        "onPremisesProvisioningErrors",
        "onPremisesSamAccountName",
        "resourceBehaviorOptions",
        "creationOptions",
        "preferredDataLocation",
    ]
    if isinstance(groups_data, list):
        groups_readable, groups_outputs = [], []
        for group_data in groups_data:
            group_readable = {camel_case_to_readable(i): j for i, j in group_data.items() if i not in fields_to_drop}
            if "@removed" in group_data:
                group_readable["Status"] = "deleted"
            groups_readable.append(group_readable)
            groups_outputs.append({k.replace(" ", ""): v for k, v in group_readable.copy().items()})

        return groups_readable, groups_outputs

    group_readable = {camel_case_to_readable(i): j for i, j in groups_data.items() if i not in fields_to_drop}
    if "@removed" in groups_data:
        group_readable["Status"] = "deleted"
    group_outputs = {k.replace(" ", ""): v for k, v in group_readable.copy().items()}

    return group_readable, group_outputs


class MsGraphClient:
    """
    Microsoft Graph Mail Client enables authorized access to a user's Office 365 mail data in a personal account.
    """

    def __init__(
        self,
        tenant_id,
        auth_id,
        enc_key,
        app_name,
        base_url,
        verify,
        proxy,
        self_deployed,
        handle_error,
        redirect_uri=None,
        auth_code=None,
        certificate_thumbprint: str | None = None,
        private_key: str | None = None,
        managed_identities_client_id: str | None = None,
    ):
        grant_type = AUTHORIZATION_CODE if auth_code and redirect_uri else CLIENT_CREDENTIALS
        resource = None if self_deployed else ""
        self.ms_client = MicrosoftClient(
            tenant_id=tenant_id,
            auth_id=auth_id,
            enc_key=enc_key,
            app_name=app_name,
            base_url=base_url,
            verify=verify,
            proxy=proxy,
            self_deployed=self_deployed,
            redirect_uri=redirect_uri,
            auth_code=auth_code,
            grant_type=grant_type,
            resource=resource,
            certificate_thumbprint=certificate_thumbprint,
            private_key=private_key,
            managed_identities_client_id=managed_identities_client_id,
            managed_identities_resource_uri=Resources.graph,
            command_prefix="msgraph-groups",
        )

        self.handle_error = handle_error

    def test_function(self):
        """Performs basic GET request to check if the API is reachable and authentication is successful.

        Returns:
            ok if successful.
        """
        self.ms_client.http_request(method="GET", url_suffix="groups", params={"$orderby": "displayName"})
        demisto.results("ok")

    def list_groups(self, order_by: str = None, next_link: str = None, top: int = None, filter_: str = None):
        """Returns all groups by sending a GET request.

        Args:
            order_by: the group fields to order by the response.
            next_link: the link for the next page of results, if exists. see Microsoft documentation for more details.
                docs.microsoft.com/en-us/graph/api/group-list?view=graph-rest-1.0
            top: sets the page size of results.
            filter_: filters results.
        Returns:
            Response from API.
        """
        if next_link:  # pagination
            return self.ms_client.http_request(method="GET", full_url=next_link)
        # default value = 100
        params = {"$top": top}
        if order_by:
            params["$orderby"] = order_by  # type: ignore
        if filter_:
            params["$filter"] = filter_  # type: ignore
        return self.ms_client.http_request(method="GET", url_suffix="groups", params=params)

    def get_group(self, group_id: str) -> dict:
        """Returns a single group by sending a GET request.

        Args:
            group_id: the group id.

        Returns:
            Response from API.
        """
        group = self.ms_client.http_request(method="GET", url_suffix=f"groups/{group_id}")
        return group

    def create_group(self, properties: dict[str, Any | None]) -> dict:
        """Create a single group by sending a POST request.

        Args:
            properties: the group properties.

        Returns:
            Response from API.
        """
        group = self.ms_client.http_request(method="POST", url_suffix="groups", json_data=properties)
        return group

    def delete_group(self, group_id: str):
        """Delete a single group by sending a DELETE request.

        Args:
            group_id: the group id to delete.
        """
        #  If successful, this method returns 204 No Content response code.
        #  It does not return anything in the response body.
        #  Using resp_type="text" to avoid parsing error in the calling method.
        self.ms_client.http_request(method="DELETE", url_suffix=f"groups/{group_id}", resp_type="text")

    def list_members(self, group_id: str, next_link: str = None, top: int = None, filter_: str = None):
        """List all group members by sending a GET request.

        Args:
            group_id: the group id to list its members.
            next_link: the link for the next page of results, if exists. see Microsoft documentation for more details.
                docs.microsoft.com/en-us/graph/api/group-list-members?view=graph-rest-1.0
            top: sets the page size of results.
            filter_: filters results.
        Returns:
            Response from API.
        """
        headers = {}

        if next_link:  # pagination
            return self.ms_client.http_request(method="GET", full_url=next_link)
        params = {"$top": top}

        if filter_:
            params["$filter"] = filter_  # type: ignore

        if count := demisto.args().get("count"):
            params["$count"] = count
            headers["ConsistencyLevel"] = "eventual"

        return self.ms_client.http_request(method="GET", url_suffix=f"groups/{group_id}/members", params=params, headers=headers)

    def add_member(self, group_id: str, properties: dict[str, str]):
        """Add a single member to a group by sending a POST request.
        Args:
            group_id: the group id to add the member to.
            properties: the member properties.
        """
        #  If successful, this method returns 204 No Content response code.
        #  It does not return anything in the response body.
        #  Using resp_type="text" to avoid parsing error in the calling method.
        self.ms_client.http_request(
            method="POST", url_suffix=f"groups/{group_id}/members/$ref", json_data=properties, resp_type="text"
        )

    def remove_member(self, group_id: str, user_id: str):
        """Remove a single member to a group by sending a DELETE request.
        Args:
            group_id: the group id to add the member to.
            user_id: the user id to remove.
        """
        #  If successful, this method returns 204 No Content response code.
        #  It does not return anything in the response body.
        #  Using resp_type="text" to avoid parsing error in the calling method.
        self.ms_client.http_request(method="DELETE", url_suffix=f"groups/{group_id}/members/{user_id}/$ref", resp_type="text")


def suppress_errors_with_404_code(func):
    def wrapper(client: MsGraphClient, args: dict):
        try:
            return func(client, args)
        except NotFoundError:
            if client.handle_error:
                human_readable = f'#### Group id -> {args.get("group_id")} does not exist'
                return human_readable, None, None
            raise

    return wrapper


def test_function_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """Performs a basic GET request to check if the API is reachable and authentication is successful.

    Args:
        client: Client object with request
        args: Usually demisto.args()

    Returns:
        Tuple.
    """
    client.test_function()
    return "ok", {}, {}


def list_groups_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """Lists all groups and return outputs in Demisto's format.

    Args:
        client: Client object with request
        args: Usually demisto.args()

    Returns:
        Outputs.
    """
    order_by = args.get("order_by")
    next_link = args.get("next_link")
    top = args.get("top")
    filter_ = args.get("filter")
    groups = client.list_groups(order_by, next_link, top, filter_)

    groups_readable, groups_outputs = parse_outputs(groups["value"])

    next_link_response = ""
    if "@odata.nextLink" in groups:
        next_link_response = groups["@odata.nextLink"]

    if next_link_response:
        entry_context = {
            f"{INTEGRATION_CONTEXT_NAME}NextLink": {"GroupsNextLink": next_link_response},
            f"{INTEGRATION_CONTEXT_NAME}(val.ID === obj.ID)": groups_outputs,
        }
        title = (
            "Groups (Note that there are more results. Please use the next_link argument to see them. The value "
            "can be found in the context under MSGraphGroupsNextLink.GroupsNextLink): "
        )
    else:
        entry_context = {f"{INTEGRATION_CONTEXT_NAME}(val.ID === obj.ID)": groups_outputs}
        title = "Groups:"

    human_readable = tableToMarkdown(
        name=title, t=groups_readable, headers=["ID", "Display Name", "Description", "Created Date Time", "Mail"], removeNull=True
    )

    return human_readable, entry_context, groups


@suppress_errors_with_404_code
def get_group_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """Get a group by group id and return outputs in Demisto's format.

    Args:
        client: Client object with request
        args: Usually demisto.args()

    Returns:
        Outputs.
    """
    group_id = str(args.get("group_id"))
    group = client.get_group(group_id)

    group_readable, group_outputs = parse_outputs(group)
    human_readable = tableToMarkdown(
        name="Groups:",
        t=group_readable,
        headers=["ID", "Display Name", "Description", "Created Date Time", "Mail", "Security Enabled", "Visibility"],
        removeNull=True,
    )
    entry_context = {f"{INTEGRATION_CONTEXT_NAME}(obj.ID === {group_id})": group_outputs}
    return human_readable, entry_context, group


def create_group_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """Create a group and return outputs in Demisto's format.

    Args:
        client: Client object with request
        args: Usually demisto.args()

    Returns:
        Outputs.
    """
    required_properties = {
        "displayName": str(args.get("display_name")),
        "mailNickname": str(args.get("mail_nickname")),
        "mailEnabled": args.get("mail_enabled") == "true",
        "securityEnabled": args.get("security_enabled"),
    }

    # create the group
    group = client.create_group(required_properties)

    # display the new group and it's properties
    group_readable, group_outputs = parse_outputs(group)
    human_readable = tableToMarkdown(
        name=f"{required_properties['displayName']} was created successfully:",
        t=group_readable,
        headers=["ID", "Display Name", "Description", "Created Date Time", "Mail", "Security Enabled", "Mail Enabled"],
        removeNull=True,
    )
    entry_context = {f"{INTEGRATION_CONTEXT_NAME}(val.ID === obj.ID)": group_outputs}
    return human_readable, entry_context, group


@suppress_errors_with_404_code
def delete_group_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """Delete a group by group id and return outputs in Demisto's format

    Args:
        client: Client object with request
        args: Usually demisto.args()

    Returns:
        Outputs.
    """
    group_id = str(args.get("group_id"))
    client.delete_group(group_id)

    # get the group data from the context
    group_data = demisto.dt(demisto.context(), f'{INTEGRATION_CONTEXT_NAME}(val.ID === "{group_id}")')
    if isinstance(group_data, list):
        group_data = group_data[0]

    # add a field that indicates that the group was deleted
    group_data["Deleted"] = True  # add a field with the members to the group
    entry_context = {f"{INTEGRATION_CONTEXT_NAME}(val.ID === obj.ID)": group_data}

    human_readable = f'Group: "{group_id}" was deleted successfully.'
    return human_readable, entry_context, NO_OUTPUTS


@suppress_errors_with_404_code
def list_members_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """List a group members by group id. return outputs in Demisto's format.

    Args:
        client: Client object with request
        args: Usually demisto.args()

    Returns:
        Outputs.
    """
    group_id = str(args.get("group_id"))
    next_link = args.get("next_link")
    top = args.get("top")
    filter_ = args.get("filter")
    members = client.list_members(group_id, next_link, top, filter_)

    if not members["value"]:
        human_readable = f"The group {group_id} has no members."
        return human_readable, NO_OUTPUTS, NO_OUTPUTS

    members_readable, members_outputs = parse_outputs(members["value"])

    # get the group data from the context
    group_data = demisto.dt(demisto.context(), f'{INTEGRATION_CONTEXT_NAME}(val.ID === "{group_id}")')
    if not group_data:
        return_error('Could not find group data in the context, please run "!msgraph-groups-get-group" to retrieve group data.')
    if isinstance(group_data, list):
        group_data = group_data[0]

    if "@odata.nextLink" in members:
        next_link_response = members["@odata.nextLink"]
        group_data["Members"] = members_outputs  # add a field with the members to the group
        group_data["MembersNextLink"] = next_link_response
        entry_context = {f"{INTEGRATION_CONTEXT_NAME}(val.ID === obj.ID)": group_data}
        title = (
            f"Group {group_id} members "
            f"(Note that there are more results. Please use the next_link argument to see them. The value can be "
            f"found in the context under {INTEGRATION_CONTEXT_NAME}.MembersNextLink): "
        )
    else:
        group_data["Members"] = members_outputs  # add a field with the members to the group
        entry_context = {f"{INTEGRATION_CONTEXT_NAME}(val.ID === obj.ID)": group_data}
        title = f"Group {group_id} members:"

    human_readable = tableToMarkdown(
        name=title, t=members_readable, headers=["ID", "Display Name", "Job Title", "Mail"], removeNull=True
    )

    return human_readable, entry_context, members


@suppress_errors_with_404_code
def add_member_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """Add a member to a group by group id and user id. return outputs in Demisto's format.

    Args:
        client: Client object with request
        args: Usually demisto.args()

    Returns:
        Outputs.
    """
    group_id = str(args.get("group_id"))
    user_id = str(args.get("user_id"))
    required_properties = {"@odata.id": f"https://graph.microsoft.com/v1.0/users/{user_id}"}
    client.add_member(group_id, required_properties)

    human_readable = f"User {user_id} was added to the Group {group_id} successfully."
    return human_readable, NO_OUTPUTS, NO_OUTPUTS


@suppress_errors_with_404_code
def remove_member_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """Remove a member from a group by group id and user id. return outputs in Demisto's format.

    Args:
        client: Client object with request
        args: Usually demisto.args()

    Returns:
        Outputs.
    """
    group_id = str(args.get("group_id"))
    user_id = str(args.get("user_id"))
    client.remove_member(group_id, user_id)

    human_readable = f'User {user_id} was removed from the Group "{group_id}" successfully.'
    return human_readable, NO_OUTPUTS, NO_OUTPUTS


def main():
    """
    PARSE AND VALIDATE INTEGRATION PARAMS
    """
    params: dict = demisto.params()
    base_url = params.get("url", "").rstrip("/") + "/v1.0/"
    tenant = params.get("creds_tenant_id", {}).get("password", "") or params.get("tenant_id") or params.get("_tenant_id")
    auth_and_token_url = params.get("creds_auth_id", {}).get("password", "") or params.get("auth_id") or params.get("_auth_id")
    enc_key = params.get("credentials", {}).get("password") or params.get("enc_key")
    verify = not params.get("insecure", False)
    redirect_uri = params.get("redirect_uri", "")
    auth_code = params.get("creds_auth_code", {}).get("password", "") or params.get("auth_code", "")
    proxy = params.get("proxy")
    handle_error: bool = argToBoolean(params.get("handle_error", "true"))
    certificate_thumbprint = params.get("credentials_certificate_thumbprint", {}).get("password", "") or params.get(
        "certificate_thumbprint"
    )
    private_key = params.get("private_key")
    managed_identities_client_id = get_azure_managed_identities_client_id(params)
    self_deployed: bool = params.get("self_deployed", False) or managed_identities_client_id is not None

    if not managed_identities_client_id:
        if not self_deployed and not enc_key:
            raise DemistoException(
                "Key must be provided. For further information see "
                "https://xsoar.pan.dev/docs/reference/articles/microsoft-integrations---authentication"
            )
        elif self_deployed and auth_code and not redirect_uri:
            raise DemistoException(
                "Please provide both Application redirect URI and Authorization code "
                "for Authorization Code flow, or None for the Client Credentials flow"
            )
        elif not enc_key and not (certificate_thumbprint and private_key):
            raise DemistoException("Key or Certificate Thumbprint and Private Key must be provided.")
        if not auth_and_token_url:
            raise Exception("Authentication ID must be provided.")
        if not tenant:
            raise Exception("Token must be provided.")

    commands = {
        "test-module": test_function_command,
        "msgraph-groups-list-groups": list_groups_command,
        "msgraph-groups-get-group": get_group_command,
        "msgraph-groups-create-group": create_group_command,
        "msgraph-groups-delete-group": delete_group_command,
        "msgraph-groups-list-members": list_members_command,
        "msgraph-groups-add-member": add_member_command,
        "msgraph-groups-remove-member": remove_member_command,
    }
    command = demisto.command()
    LOG(f"Command being called is {command}")

    try:
        client: MsGraphClient = MsGraphClient(
            tenant_id=tenant,
            auth_id=auth_and_token_url,
            enc_key=enc_key,
            app_name=APP_NAME,
            base_url=base_url,
            verify=verify,
            proxy=proxy,
            self_deployed=self_deployed,
            redirect_uri=redirect_uri,
            auth_code=auth_code,
            handle_error=handle_error,
            certificate_thumbprint=certificate_thumbprint,
            private_key=private_key,
            managed_identities_client_id=managed_identities_client_id,
        )
        if command == "msgraph-groups-generate-login-url":
            return_results(generate_login_url(client.ms_client))
        elif command == "msgraph-groups-auth-reset":
            return_results(reset_auth())
        else:
            human_readable, entry_context, raw_response = commands[command](client, demisto.args())  # type: ignore
            return_outputs(readable_output=human_readable, outputs=entry_context, raw_response=raw_response)

    except Exception as err:
        return_error(str(err))


if __name__ in ["__main__", "builtin", "builtins"]:
    main()