Microsoft Graph Calendar

import urllib3
from CommonServerPython import *
from MicrosoftApiModule import *  # noqa: E402

# Disable insecure warnings
urllib3.disable_warnings()

INTEGRATION_CONTEXT_NAME = "MSGraphCalendar"
DEFAULT_PAGE_SIZE = 100
NO_OUTPUTS: dict = {}
APP_NAME = "ms-graph-calendar"
EVENT_HEADERS = ["Subject", "Organizer", "Attendees", "Start", "End", "ID"]
CALENDAR_HEADERS = ["Name", "Owner Name", "Owner Address", "ID"]


def camel_case_to_readable(cc: str | dict, fields_to_drop: list[str] = None) -> str | dict:
    """
    'camelCase' -> 'Camel Case' (text or dictionary keys)

    Args:
        cc: either a dictionary or a text to transform
        fields_to_drop: keys to drop from input dictionary

    Returns:
        A Camel Cased string of Dict.
    """
    if fields_to_drop is None:
        fields_to_drop = []
    if isinstance(cc, str):
        if cc == "id":
            return "ID"
        return "".join(" " + char if char.isupper() else char.strip() for char in cc).strip().title()

    elif isinstance(cc, dict):
        return {camel_case_to_readable(field): value for field, value in cc.items() if field not in fields_to_drop}
    return cc


def snakecase_to_camelcase(sc: str | dict, fields_to_drop: list[str] = None) -> str | dict:
    """
    'snake_case' -> 'snakeCase' (text or dictionary keys)

    Args:
        sc: either a dictionary or a text to transform
        fields_to_drop: keys to drop from input dictionary

    Returns:
        A connectedCamelCased string of Dict.
    """
    if fields_to_drop is None:
        fields_to_drop = []
    if isinstance(sc, str):
        return "".join([word.title() for word in sc.split("_")])

    elif isinstance(sc, dict):
        return {snakecase_to_camelcase(field): value for field, value in sc.items() if field not in fields_to_drop}
    return sc


def parse_events(raw_events: dict | list[dict]) -> tuple[list[dict], list[dict]]:
    """
    Parse Calendar Events json data coming from Microsoft Graph into Demisto readable format
    :param raw_events: raw events data
    """
    # Fields to filter, dropping to not bloat the incident context.
    fields_to_drop = ["@odata.etag", "color"]
    if not isinstance(raw_events, list):
        raw_events = [raw_events]

    readable_events, context_output = [], []
    for event in raw_events:
        event_readable: dict = camel_case_to_readable(event, fields_to_drop)  # type: ignore
        if "@removed" in event:
            event_readable["Status"] = "deleted"
        event_context = {field.replace(" ", ""): value for field, value in event_readable.items()}

        event_readable = {
            "Subject": event_readable.get("Subject"),
            "ID": event_readable.get("ID"),
            "Organizer": demisto.get(event_readable, "Organizer.emailAddress.name"),
            "Attendees": [att.get("emailAddress", {}).get("name") for att in event_readable.get("Attendees", [])],
            "Start": event_readable.get("Start", {}).get("dateTime"),
            "End": event_readable.get("End", {}).get("dateTime"),
        }
        readable_events.append(event_readable)
        context_output.append(event_context)

    return readable_events, context_output


def parse_calendar(raw_calendars: dict | list[dict]) -> tuple[list[dict], list[dict]]:
    """
    Parse Calendar json data coming from Microsoft Graph into Demisto readable format
    :param raw_calendars: raw calendars data
    """
    if not isinstance(raw_calendars, list):
        raw_calendars = [raw_calendars]

    readable_calendars, context_output = [], []
    for raw_calendar in raw_calendars:
        readable_calendar: dict = camel_case_to_readable(raw_calendar, ["@odata.context", "color"])  # type: ignore
        if "@removed" in readable_calendar:
            readable_calendar["Status"] = "deleted"
        context_calendar = {field.replace(" ", ""): value for field, value in readable_calendar.items()}

        readable_calendar = {
            "Name": readable_calendar.get("Name"),
            "Owner Name": readable_calendar.get("Owner", {}).get("name"),
            "Owner Address": readable_calendar.get("Owner", {}).get("address"),
            "ID": readable_calendar.get("ID"),
        }
        context_output.append(context_calendar)
        readable_calendars.append(readable_calendar)

    return readable_calendars, context_output


def process_event_params(
    body: str = "", start: str = "", end: str = "", time_zone: str = "", attendees: str = "", location: str = "", **other_params
) -> dict:
    # some parameters don't need any processing
    event_params: dict[str, str | dict | list[dict]] = other_params

    event_params["body"] = {"content": body}
    event_params["location"] = {"displayName": location}
    if start:
        event_params["start"] = {"dateTime": start, "timeZone": time_zone}
    if end:
        event_params["end"] = {"dateTime": end, "timeZone": time_zone}
    event_params["attendees"] = [{"emailAddress": {"address": attendee}} for attendee in attendees.split(",")]
    return event_params


class MsGraphClient:
    def __init__(
        self,
        tenant_id,
        auth_id,
        enc_key,
        app_name,
        base_url,
        verify,
        proxy,
        default_user,
        self_deployed,
        certificate_thumbprint,
        private_key,
        managed_identities_client_id,
    ):
        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,
            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-calendar",
        )

        self.default_user = default_user

    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="users/")
        return "ok", NO_OUTPUTS, NO_OUTPUTS

    def get_calendar(self, user: str, calendar_id: str = None) -> dict:
        """Returns a single calendar by sending a GET request.

        Args:
        :argument user: the user id | userPrincipalName
        :argument calendar_id: calendar id  | name
        """
        if not user and not self.default_user:
            return_error(
                "No user was provided. Please make sure to enter the use either in the instance setting,"
                " or in the command parameter."
            )
        calendar_raw = self.ms_client.http_request(
            method="GET", url_suffix=f"users/{user}/calendar" + f"s/{calendar_id}" if calendar_id else ""
        )

        return calendar_raw

    def list_calendars(
        self, user: str, order_by: str = None, next_link: str = None, top: int = DEFAULT_PAGE_SIZE, filter_by: str = None
    ) -> dict:
        """
        Lists all calendars by sending a GET request.

        Args:
        :argument user: the user id | userPrincipalName
        :argument order_by: specify the sort order of the items returned from Microsoft Graph
        :argument next_link: link for the next page of results, if exists. See Microsoft documentation for more details.
            docs.microsoft.com/en-us/graph/api/event-list?view=graph-rest-1.0
        :argument top: specify the page size of the result set.
        filter_by: filters results.
        """
        params = {"$orderby": order_by} if order_by else {}
        if next_link:  # pagination
            calendars = self.ms_client.http_request(method="GET", full_url=next_link)
        elif filter_by:
            calendars = self.ms_client.http_request(
                method="GET", url_suffix=f"users/{user}/calendars?$filter={filter_by}&$top={top}", params=params
            )
        else:
            calendars = self.ms_client.http_request(method="GET", url_suffix=f"users/{user}/calendars?$top={top}", params=params)

        return calendars

    def list_events(
        self,
        user: str,
        calendar_id: str = "",
        order_by: str = None,
        next_link: str = None,
        top: int = DEFAULT_PAGE_SIZE,
        filter_by: str = None,
    ) -> dict:
        """
        Returns all events by sending a GET request.

        Args:
        :argument user: the user id | userPrincipalName
        :argument calendar_id: calendar id  | name
        :argument order_by: specify the sort order of the items returned from Microsoft Graph
        :argument next_link: the link for the next page of results. see Microsoft documentation for more details.
        :argument top: specify the page size of the result set.
        :argument filter_by: filters results.
        """
        calendar_url = f"{user}/calendars/{calendar_id}" if calendar_id else user
        params = {"$orderby": order_by} if order_by else {}
        if next_link:  # pagination
            events = self.ms_client.http_request(method="GET", full_url=next_link)
        elif filter_by:
            events = self.ms_client.http_request(
                method="GET", url_suffix=f"users/{calendar_url}/events?$filter={filter_by}&$top={top}", params=params
            )
        else:
            events = self.ms_client.http_request(
                method="GET", url_suffix=f"users/{calendar_url}/events?$top={top}", params=params
            )
        return events

    def get_event(self, user: str, event_id: str) -> dict:
        """
        Create a single event in a user calendar, or the default calendar of an Office 365 group.

        Args:
        :argument user: the user id | userPrincipalName
        :argument event_id: the event id
        """
        event = self.ms_client.http_request(method="GET", url_suffix=f"users/{user}/calendar/events/{event_id}")

        return event

    def create_event(self, user: str, calendar_id: str = "", **kwargs) -> dict:
        """
        Create a single event in a user calendar, or the default calendar of an Office 365 group.

        Args:
        :argument user: the user id | userPrincipalName
        :argument calendar_id: calendar id  | name

        Event Properties:
        :keyword attendees: The collection of attendees for the event.
        :keyword body: The body of the message associated with the event. It can be in HTML or text format.
        :keyword subject: The text of the event's subject line.
        :keyword location: The location of the event. an event as an online meeting such as a Zoom meeting. Read-only.
        :keyword end: The date, time, and time zone that the event ends. By default, the end time is in UTC.
        :keyword originalEndTimeZone: The end time zone that was set when the event was created.
        :keyword originalStart: The Timestamp type represents date and time using ISO 8601 format in UTC time.
         For example, midnight UTC on Jan 1, 2014 would look like this: '2014-01-01T00:00:00Z'
        :keyword originalStartTimeZone: The start time zone that was set when the event was created.
        """
        if calendar_id:
            event = self.ms_client.http_request(
                method="POST", url_suffix=f"/users/{user}/calendars/{calendar_id}/events", json_data=kwargs
            )
        else:
            event = self.ms_client.http_request(method="POST", url_suffix=f"users/{user}/calendar/events", json_data=kwargs)
        return event

    def update_event(self, user: str, event_id: str, **kwargs) -> dict:
        """
        Create a single event in a user calendar, or the default calendar of an Office 365 group.

        Args:
        :argument user: the user id | userPrincipalName
        :argument event_id: the event ID

        Event Properties:
        :keyword attendees: The collection of attendees for the event.
        :keyword body: The body of the message associated with the event. It can be in HTML or text format.
        :keyword subject:The text of the event's subject line.
        :keyword location: The location of the event.
         an event as an online meeting such as a Skype meeting. Read-only.
        :keyword end: The date, time, and time zone that the event ends. By default, the end time is in UTC.
        :keyword originalEndTimeZone: The end time zone that was set when the event was created.
        :keyword originalStart: The Timestamp type represents date and time using ISO 8601 format in UTC time.
         For example, midnight UTC on Jan 1, 2014 would look like this: '2014-01-01T00:00:00Z'
        :keyword originalStartTimeZone: The start time zone that was set when the event was created.
        """
        event = self.ms_client.http_request(
            method="PATCH", url_suffix=f"users/{user}/calendar/events/{event_id}", json_data=kwargs
        )
        return event

    def delete_event(self, user: str, event_id: str):
        """
        Delete a single event by sending a DELETE request.

        Args:
        :argument user: the user id | userPrincipalName
        :argument id: the event id
        """
        #  If successful, this method returns 204 No Content response code.
        #  It does not return anything in the response body.
        self.ms_client.http_request(method="DELETE", url_suffix=f"users/{user}/calendar/events/{event_id}", resp_type="text")


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

    Args:
        client: Client object with request
        args: Usually demisto.args()
    """
    events = client.list_events(**args)

    events_readable, events_outputs = parse_events(events.get("value"))  # type: ignore

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

    if next_link_response:
        entry_context = {
            f"{INTEGRATION_CONTEXT_NAME}.Event(val.ID === obj.ID).NextLink": next_link_response,
            f"{INTEGRATION_CONTEXT_NAME}.Event(val.ID === obj.ID)": events_outputs,
        }
        title = "Events (Note that there are more results. Please use the next_link argument to see them.):"
    else:
        entry_context = {f"{INTEGRATION_CONTEXT_NAME}.Event(val.ID === obj.ID)": events_outputs}
        title = "Events:"

    human_readable = tableToMarkdown(name=title, t=events_readable, headers=EVENT_HEADERS, removeNull=True)

    return human_readable, entry_context, events


def get_event_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """
    Retrieves an event by event id and return outputs in Demisto's format

    Args:
        client: Client object with request
        args: Usually demisto.args()
    """
    event = client.get_event(**args)

    # display the event and it's properties
    event_readable, event_outputs = parse_events(event)
    human_readable = tableToMarkdown(
        name=f"Event - {event_outputs[0].get('Subject')}", t=event_readable, headers=EVENT_HEADERS, removeNull=True
    )
    entry_context = {f"{INTEGRATION_CONTEXT_NAME}.Event(val.ID === obj.ID)": event_outputs}
    return human_readable, entry_context, event


def create_event_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """
    Creates an event by event id and return outputs in Demisto's format

    Args:
        client: Client object with request
        args: Usually demisto.args()
    """
    args = process_event_params(**args)
    params: dict = snakecase_to_camelcase(args, fields_to_drop=["user", "calendar_id"])  # type: ignore

    # create the event
    event = client.create_event(user=args.get("user", ""), calendar_id=args.get("calendar_id", ""), **params)

    # display the new event and it's properties
    event_readable, event_outputs = parse_events(event)
    human_readable = tableToMarkdown(
        name="Event was created successfully:", t=event_readable, headers=EVENT_HEADERS, removeNull=True
    )
    entry_context = {f"{INTEGRATION_CONTEXT_NAME}.Event(val.ID === obj.ID)": event_outputs}
    return human_readable, entry_context, event


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

    Args:
        client: Client object with request
        args: Usually demisto.args()
    """
    event_id = args.get("event_id", "")
    args = process_event_params(**args)
    params: dict = snakecase_to_camelcase(args, fields_to_drop=["user", "calendar_id", "event_id"])  # type: ignore

    # update the event
    event = client.update_event(user=args.get("user", ""), event_id=args.get("event_id", ""), **params)

    # display the updated event and it's properties
    event_readable, event_outputs = parse_events(event)
    human_readable = tableToMarkdown(name="Event:", t=event_readable, headers=EVENT_HEADERS, removeNull=True)
    entry_context = {f"{INTEGRATION_CONTEXT_NAME}(obj.ID === {event_id})": event_outputs}
    return human_readable, entry_context, event


def delete_event_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """
    Delete an event by event id and return outputs in Demisto's format

    Args:
        client: Client object with request
        args: Usually demisto.args()
    """
    event_id = str(args.get("event_id"))
    client.delete_event(**args)

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

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

    human_readable = "Event was deleted successfully."
    return human_readable, entry_context, NO_OUTPUTS


def list_calendars_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """
    Get all the user's calendars (/calendars navigation property)

    Args:
        client: Client object with request
        args: Usually demisto.args()
    """
    calendar = client.list_calendars(**args)

    calendar_readable, calendar_outputs = parse_calendar(calendar.get("value"))  # type: ignore

    entry_context = {f"{INTEGRATION_CONTEXT_NAME}.Calendar(val.ID === obj.ID)": calendar_outputs}
    title = "Calendar:"

    human_readable = tableToMarkdown(name=title, t=calendar_readable, headers=CALENDAR_HEADERS, removeNull=True)

    return human_readable, entry_context, calendar


def get_calendar_command(client: MsGraphClient, args: dict) -> tuple[str, dict, dict]:
    """
    Get the properties and relationships of a calendar object.
    The calendar can be one for a user, or the default calendar of an Office 365 group.

    Args:
        client: Client object with request
        args: Usually demisto.args()
    """
    calendar = client.get_calendar(**args)

    calendar_readable, calendar_outputs = parse_calendar(calendar)

    entry_context = {f"{INTEGRATION_CONTEXT_NAME}.Calendar(val.ID === obj.ID)": calendar_outputs}
    title = "Calendar:"

    human_readable = tableToMarkdown(name=title, t=calendar_readable, headers=CALENDAR_HEADERS, removeNull=True)

    return human_readable, entry_context, calendar


def module_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
    """
    return client.test_function()


def main():
    params: dict = demisto.params()
    url = params.get("url", "").rstrip("/") + "/v1.0/"
    tenant = params.get("credentials_tenant_id", {}).get("password") or params.get("tenant_id")
    auth_and_token_url = params.get("credentials_auth_id", {}).get("password") or params.get("auth_id", "")
    enc_key = params.get("credentials_enc_key", {}).get("password") or params.get("enc_key")
    verify = not params.get("insecure", False)
    proxy = params.get("proxy", False)
    default_user = params.get("default_user")
    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 not enc_key and not (certificate_thumbprint and private_key):
            raise DemistoException("Key or Certificate Thumbprint and Private Key must be provided.")

    commands = {
        "test-module": module_test_function_command,
        "msgraph-calendar-list-calendars": list_calendars_command,
        "msgraph-calendar-get-calendar": get_calendar_command,
        "msgraph-calendar-list-events": list_events_command,
        "msgraph-calendar-get-event": get_event_command,
        "msgraph-calendar-create-event": create_event_command,
        "msgraph-calendar-update-event": update_event_command,
        "msgraph-calendar-delete-event": delete_event_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=url,
            verify=verify,
            proxy=proxy,
            default_user=default_user,
            self_deployed=self_deployed,
            certificate_thumbprint=certificate_thumbprint,
            private_key=private_key,
            managed_identities_client_id=managed_identities_client_id,
        )
        if "user" not in demisto.args():
            demisto.args()["user"] = client.default_user
        if command == "msgraph-calendar-auth-reset":
            return_results(reset_auth())
        else:
            # Run the command
            human_readable, entry_context, raw_response = commands[command](client, demisto.args())  # type: ignore
            # create a war room entry
            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()