GenesysCloud

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401
import aiohttp
from http import HTTPStatus
import asyncio
import math
from typing import Any
from datetime import datetime, timedelta, UTC

""" CONSTANTS """

# Dataset
DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
VENDOR = "Genesys"
PRODUCT = "Cloud"

# Access Token
ACCESS_TOKEN_KEY = "access_token"
TOKEN_TYPE_KEY = "token_type"
TOKEN_TTL_KEY = "expires_in"  # TTL in seconds
TOKEN_ERROR_KEY = "error"  # Optional error message if request fails
TOKEN_VALID_UNTIL_KEY = "valid_until"

# Default Values
DEFAULT_SERVER_URL = "https://api.mypurecloud.com"
DEFAULT_SERVICE_NAMES = [
    "Architect",
    "PeoplePermissions",
    "ContactCenter",
    "Groups",
    "Telephony",
    "Outbound",
    "Routing",
    "Integrations",
    "AnalyticsReporting",
]
DEFAULT_AUDIT_PAGE_SIZE = 500
DEFAULT_AUDIT_RETRY_COUNT = 3
DEFAULT_GET_EVENTS_LIMIT = 10
DEFAULT_FETCH_EVENTS_LIMIT = 2500  # per service
DEFAULT_TOKEN_TTL = 86400
DEFAULT_TOKEN_TYPE = "bearer"

""" CLIENT CLASS """


class AsyncClient:
    """An asynchronous client for interacting with the Genesys Cloud API; used for SIEM event collection"""

    def __init__(self, base_url: str, client_id: str, client_secret: str, verify: bool, proxy: bool):
        self.base_url = base_url
        self._client_id = client_id
        self._client_secret = client_secret
        self._access_token: str | None = None
        self._verify = verify
        self._proxy_url = handle_proxy().get("http", "") if proxy else None

    async def __aenter__(self):
        self._session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=self._verify), proxy=self._proxy_url)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            exception_traceback = "".join(traceback.format_exception(exc_type, exc_val, exc_tb))
            demisto.error(f"AsyncClient context exited with an exception: {exception_traceback}.")
        else:
            demisto.debug("AsyncClient context exited normally.")

        # Always ensure HTTP client session is closed
        await self._session.close()

    async def _generate_new_access_token(self) -> dict[str, Any]:
        """
        Generates a new OAuth2 access token using client credentials flow.

        Raises:
            ClientResponseError: If request failed.
            DemistoException: If response contains an error message and/or no access token.

        Returns:
            dict[str, Any]: The token raw API response.

        Example:
            >>> async_client.generate_access_token()
            {
                "access_token": "token",
                "token_type": "bearer",
                "expires_in": 86400,
                "error": "optional-error-message",
            }
        """
        login_url = self.base_url.replace("api.", "login.")
        token_url = urljoin(login_url, "/oauth/token")

        demisto.debug(f"Requesting new OAuth2 access token using {token_url=}.")
        async with self._session.post(
            url=token_url,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            params={"grant_type": "client_credentials"},
            auth=aiohttp.BasicAuth(self._client_id, self._client_secret),
        ) as response:
            response_json = await response.json()
            try:
                response.raise_for_status()
            except aiohttp.ClientResponseError as e:
                raise DemistoException(
                    f"Request to {token_url} failed with HTTP {e.status} status. Got response: {response_json}."
                ) from e

            if error_message := response_json.get(TOKEN_ERROR_KEY):  # Optional error message if request fails
                raise DemistoException(f"Request to {token_url} failed. Got error: {error_message}.")

            if not response_json.get(ACCESS_TOKEN_KEY):
                raise DemistoException(f"Request to {token_url} failed. Failed to get access token from response.")

            token_type = response_json.get(TOKEN_TYPE_KEY)
            token_ttl = response_json.get(TOKEN_TTL_KEY)  # TTL in seconds

            demisto.debug(f"Successfully obtained OAuth2 access token. {token_type=}, {token_ttl=}.")
            return response_json

    async def get_authorization_header(self, force_generate_new_token: bool = False) -> str:
        """
        Constructs Authorization header using the access token in the integration context (if found), or generating a new one.

        Args:
            force_generate_new_token (bool, optional): Whether to request a new OAuth access token. Defaults to False.

        Returns:
            str: The Authorization header containing the token type and access token.

        Example:
            >>> async_client.get_authorization_header()
            "Bearer MyToken1245"
        """
        demisto.debug(f"Constructing Authorization header using {force_generate_new_token=}.")
        integration_context = get_integration_context()
        access_token = integration_context.get(ACCESS_TOKEN_KEY)
        token_type = integration_context.get(TOKEN_TYPE_KEY, DEFAULT_TOKEN_TYPE)
        token_valid_until = arg_to_datetime(integration_context.get(TOKEN_VALID_UNTIL_KEY))
        is_valid_token = token_valid_until and token_valid_until > datetime.now(tz=UTC)
        demisto.debug(f"Found in integration context {token_valid_until=}, {is_valid_token=}.")

        if access_token and is_valid_token and not force_generate_new_token:
            demisto.debug("Using valid access token in integration context to construct Authorization header.")
        else:
            demisto.debug("Generating new access token.")
            token_response = await self._generate_new_access_token()
            access_token = token_response.get(ACCESS_TOKEN_KEY)
            token_type = token_response.get(TOKEN_TYPE_KEY, DEFAULT_TOKEN_TYPE)
            token_ttl = token_response.get(TOKEN_TTL_KEY, DEFAULT_TOKEN_TTL) - 300  # subtract 5 minutes as a safety margin
            token_response[TOKEN_VALID_UNTIL_KEY] = (datetime.now(tz=UTC) + timedelta(seconds=token_ttl)).isoformat()
            demisto.debug("Saving new access token in integration context.")
            set_integration_context(token_response)

        demisto.debug(f"Constructed Authorization header using {token_type=}.")
        return f"{token_type.capitalize()} {access_token}"

    async def _send_audits_post_request(
        self,
        url: str,
        params: dict[str, str],
        headers: dict[str, str],
        body: dict[str, Any],
    ) -> dict[str, Any]:
        """
        Sends HTTP POST request to get realtime audits.

        Args:
            url (str): The full URL.
            params (dict[str, str]): The URL query parameters.
            headers (dict[str, str]): The request headers, including the "Authorization" header.
            body (dict[str, Any]): The request body, including "interval", "serviceName", and "pageNumber".

        Raises:
            ClientResponseError: If the request fails.

        Returns:
            dict[str, Any]: The audit events raw API response.
        """
        async with self._session.post(url=url, params=params, headers=headers, json=body) as response:
            response.raise_for_status()
            return await response.json()

    async def get_realtime_audits(
        self,
        from_date: str,
        to_date: str,
        service_name: str,
        page_number: int,
        page_size: int = DEFAULT_AUDIT_PAGE_SIZE,
        max_retries: int = DEFAULT_AUDIT_RETRY_COUNT,
    ) -> dict[str, Any]:
        """
        Retrieves audit events from Genesys Cloud for a specific service using the realtime audit query API.

        Args:
            from_date (str): The start date for the audit events in ISO 8601 format.
            to_date (str): The end date for the audit events in ISO 8601 format.
            service_name (str): The name of the service to fetch events for.
            page_number (int): The page number to retrieve.
            page_size (int): The number of items per page. Defaults to 500.
            max_retries (int): Maximum number of retries for rate limit errors. Defaults to 3.

        Raises:
            ClientResponseError: If the request fails (after retry on 401 or 429 errors or some other status code).

        Returns:
            dict[str, Any]: A dictionary containing the audit events raw API response.
        """
        url = urljoin(self.base_url, "/api/v2/audits/query/realtime")
        params = {"expand": "user"}
        body = {
            "interval": f"{from_date}/{to_date}",
            "serviceName": service_name,
            "sort": [{"name": "Timestamp", "sortOrder": "ascending"}],
            "pageNumber": page_number,
            "pageSize": page_size,
        }
        # Get authorization header (will try to get token from integration context or generate new one if needed)
        headers = {"Content-Type": "application/json", "Authorization": await self.get_authorization_header()}

        demisto.debug(f"[{service_name}] Requesting audits using {from_date=}, {to_date=}, {page_number=}.")

        retry_count = 0
        while retry_count <= max_retries:
            try:
                response_json = await self._send_audits_post_request(url, params=params, headers=headers, body=body)
                break  # Success, exit retry loop

            except aiohttp.ClientResponseError as e:
                status_code = e.status
                error_message = e.message

                if status_code == HTTPStatus.UNAUTHORIZED:
                    demisto.debug(
                        f"[{service_name}] Received HTTP 401 Unauthorized error. Forcing new token generation and retrying..."
                    )
                    headers["Authorization"] = await self.get_authorization_header(force_generate_new_token=True)
                    response_json = await self._send_audits_post_request(url, params=params, headers=headers, body=body)
                    break  # Success after token refresh, exit retry loop

                if status_code == HTTPStatus.TOO_MANY_REQUESTS:
                    if retry_count < max_retries:
                        # Exponential backoff: 2^retry_count seconds (1s, 2s, 4s)
                        wait_time = 2**retry_count
                        demisto.debug(
                            f"[{service_name}] Received HTTP 429 Too Many Requests error. "
                            f"Retrying {retry_count + 1} of {max_retries} after {wait_time} seconds..."
                        )
                        await asyncio.sleep(wait_time)
                        retry_count += 1

                else:
                    demisto.error(
                        f"[{service_name}] Request using {from_date=}, {to_date=}, {page_number=} failed. "
                        f"Got {status_code=}, {error_message=}."
                    )
                    raise DemistoException(f"Request to {e.request_info.url} failed with HTTP {e.status} status.") from e

        entities_count = len(response_json.get("entities", []))
        demisto.debug(f"[{service_name}] Fetched {entities_count} audits using {from_date=}, {to_date=}, {page_number=}.")

        return response_json


""" HELPER FUNCTIONS """


def deduplicate_and_format_events(
    raw_response: dict[str, Any],
    all_fetched_ids: set[str],
    service_name: str,
) -> list[dict[str, Any]]:
    """
    Processes events from a raw API response, deduplicates them, and adds the `_time` and `source_log_type` fields.

    Args:
        raw_response (dict[str, Any]): A dictionary containing the raw API response of the audit events.
        all_fetched_ids (set[str]): A set of event IDs that have already been fetched.
        service_name (str): The name of the service the events were fetched from.

    Returns:
        list[dict[str, Any]]: A list of new, processed events.
    """
    events = []
    for event in raw_response.get("entities", []):
        event_id = event.get("id")
        if event_id in all_fetched_ids:
            demisto.debug(f"Skipping duplicate {event_id=}.")
            continue
        all_fetched_ids.add(event_id)
        event["_time"] = arg_to_datetime(event.get("eventDate"), required=True).strftime(DATE_FORMAT)  # type: ignore [union-attr]
        event["source_log_type"] = service_name
        events.append(event)
    return events


async def get_audit_events_for_service(
    client: AsyncClient,
    from_date: str,
    to_date: str,
    service_name: str,
    limit: int,
    last_fetched_ids: list[str] | None = None,
) -> list[dict[str, Any]]:
    """
    Asynchronously fetches audit events from Genesys Cloud for a specific service.

    Args:
        client (AsyncClient): An instance of the AsyncClient.
        from_date (str): The start date for the audit events in ISO 8601 format.
        to_date (str): The end date for the audit events in ISO 8601 format.
        service_name (str): The name of the service to fetch events for.
        limit (int): The maximum number of events to retrieve.
        last_fetched_ids (list[str]): A list of IDs of events that have already been fetched.

    Returns:
        list[dict[str, Any]]: A list of new audit events for the service.
    """
    last_fetched_ids = last_fetched_ids or []
    all_fetched_ids = set(last_fetched_ids)
    all_events: list[dict[str, Any]] = []

    # Calculate number of pages needed based on limit and page size
    start_page_number = 1  # Page numbers start from 1
    stop_page_number = math.ceil(limit / DEFAULT_AUDIT_PAGE_SIZE) + 1  # Stop page not included (range stops one before)
    demisto.debug(f"[{service_name}] Fetching {stop_page_number} pages concurrently to retrieve up to {limit} events.")

    # Create tasks for fetching all pages concurrently
    page_tasks = [
        client.get_realtime_audits(
            from_date=from_date,
            to_date=to_date,
            service_name=service_name,
            page_number=page_number,
            page_size=DEFAULT_AUDIT_PAGE_SIZE,
        )
        for page_number in range(start_page_number, stop_page_number)  # Page numbers start from 1
    ]

    # Fetch all pages concurrently
    page_responses = await asyncio.gather(*page_tasks)

    # Process results from all pages
    for page_number, page_response in enumerate(page_responses, start=start_page_number):
        # Process and deduplicate events from this page
        page_events = deduplicate_and_format_events(page_response, all_fetched_ids, service_name)
        for event in page_events:
            all_events.append(event)

            # Stop if limit was reached
            if len(all_events) >= limit:
                demisto.debug(f"[{service_name}] Reached {limit=} after processing events on {page_number=}.")
                break

    demisto.debug(f"[{service_name}] Fetched total of {len(all_events)} events from {stop_page_number} pages.")
    return all_events


""" COMMAND FUNCTIONS """


async def test_module(client: AsyncClient, service_names: list[str]) -> str:
    """
    Tests the connection to the Genesys Cloud realtime audit events API.

    Args:
        client (AsyncClient): An instance of the AsyncClient.
        service_names (list[str]): List of service names to fetch events from.

    Returns:
        str: "ok" if connection to the realtime audit events API succeeded.
    """
    await fetch_events_command(client, last_run={}, max_fetch=1, service_names=service_names)
    return "ok"


async def get_events_command(client: AsyncClient, args: dict[str, Any]) -> tuple[list[dict[str, Any]], CommandResults]:
    """
    Implements the `genesis-cloud-get-events` command. Gets audit events using the AsyncClient.

    Args:
        client (AsyncClient): An instance of the AsyncClient.
        args (dict[str, Any]): The command arguments.

    Returns:
        tuple[list[dict[str, Any]], CommandResults]: A tuple of the events list and the CommandResults.
    """
    from_date = arg_to_datetime(args.get("from_date")) or (datetime.now(tz=UTC) - timedelta(hours=1))
    to_date = arg_to_datetime(args.get("to_date")) or datetime.now(tz=UTC)
    limit = arg_to_number(args.get("limit")) or DEFAULT_GET_EVENTS_LIMIT
    service_name = args["service_name"]

    events = await get_audit_events_for_service(
        client=client,
        from_date=from_date.strftime(DATE_FORMAT),
        to_date=to_date.strftime(DATE_FORMAT),
        service_name=service_name,
        limit=limit,
    )

    human_readable = tableToMarkdown(name=f"Genesys Cloud Audit Events from Service: {service_name}", t=events)
    return events, CommandResults(readable_output=human_readable)


async def fetch_events_command(
    client: AsyncClient,
    last_run: dict,
    max_fetch: int,
    service_names: list[str],
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    """
    Implements `fetch-events` command. Fetches audit events using the AsyncClient.

    Fetches events from multiple services concurrently. If one service fails,
    it will not impact the fetching of events from other services.

    Args:
        client (AsyncClient): An instance of the AsyncClient.
        last_run (dict): The last run object.
        max_fetch (int): The maximum number of events to fetch per service.
        service_names (list[str]): List of service names to fetch events from.

    Returns:
        tuple[dict[str, Any], list[dict[str, Any]]]: A tuple of the next run object and a list of fetched events.
    """
    demisto.debug(f"Starting fetching events with {last_run=}.")
    default_from_date = (datetime.now(tz=UTC) - timedelta(minutes=1)).strftime(DATE_FORMAT)
    to_date = datetime.now(tz=UTC).strftime(DATE_FORMAT)

    service_tasks = []
    # Create tasks for fetching events from each service concurrently
    for service_name in service_names:
        service_last_run = last_run.get(service_name, {})
        from_date = service_last_run.get("from_date") or default_from_date
        last_fetched_ids = service_last_run.get("last_fetched_ids", [])

        service_tasks.append(
            get_audit_events_for_service(
                client=client,
                from_date=from_date,
                to_date=to_date,
                service_name=service_name,
                limit=max_fetch,
                last_fetched_ids=last_fetched_ids,
            )
        )

    # Gather results with `return_exceptions=True` to prevent one failure from affecting others
    demisto.debug(f"Fetching events from {len(service_names)} services concurrently: {service_names}")
    results = await asyncio.gather(*service_tasks, return_exceptions=True)

    # Process results and handle any exceptions
    next_run = {}
    all_events = []
    per_service_errors = {}

    for service_name, result in zip(service_names, results):
        service_last_run = last_run.get(service_name, {})

        if isinstance(result, Exception):
            # Log the error but continue processing events from other services
            service_traceback = "".join(traceback.format_exception(type(result), result, result.__traceback__))
            demisto.error(f"[{service_name}] Failed to fetch events. Traceback: {service_traceback}.")
            per_service_errors[service_name] = str(result)

        elif isinstance(result, list):
            # Successfully fetched events from service
            service_events = result
            demisto.debug(f"[{service_name}] Fetched {len(service_events)} events using {from_date=}, {to_date=}, {max_fetch=}.")

            # If no new events from service, set its next run same as last run
            if not service_events:
                demisto.debug(f"[{service_name}] No new events found since {service_last_run=}.")
                next_run[service_name] = service_last_run
                continue

            # Get the newest event timestamp
            newest_event_time = service_events[-1].get("eventDate")
            demisto.debug(f"[{service_name}] Got {len(service_events)} deduplicated events with {newest_event_time=}.")

            # Get the IDs of the service events that have the newest time
            new_last_fetched_ids = [event.get("id") for event in service_events if event.get("eventDate") == newest_event_time]

            # Update next run for service
            service_next_run = {"from_date": newest_event_time, "last_fetched_ids": new_last_fetched_ids}
            demisto.debug(f"[{service_name}] Updating {service_next_run=} after fetching {len(service_events)} events.")

            # Set next run
            all_events.extend(service_events)
            next_run[service_name] = service_next_run

        else:
            # Unlikely case of getting unknown / unexpected result
            demisto.debug(f"[{service_name}] Unexpected result type: {type(result)}.")

    # If all services failed, raise an exception
    if len(per_service_errors) == len(service_names):
        error_summary = "\n".join(f"{service_name}: {error}" for service_name, error in per_service_errors.items())
        raise DemistoException(f"Fetching events failed from all services:\n{error_summary}.")

    demisto.debug(f"Finished fetching {len(all_events)} events. Setting {next_run=}.")
    return next_run, all_events


""" MAIN FUNCTION """


async def main() -> None:  # pragma: no cover
    params: dict[str, Any] = demisto.params()
    args: dict[str, Any] = demisto.args()
    command: str = demisto.command()

    # HTTP Connection
    base_url: str = params.get("url") or DEFAULT_SERVER_URL
    verify: bool = not params.get("insecure", False)
    proxy: bool = params.get("proxy", False)

    # OAuth Credentials
    client_id: str = params.get("credentials", {}).get("identifier", "")
    client_secret: str = params.get("credentials", {}).get("password", "")

    # Fetch Events
    max_fetch: int = arg_to_number(params.get("max_fetch")) or DEFAULT_FETCH_EVENTS_LIMIT
    service_names: list = argToList(params.get("service_names")) or DEFAULT_SERVICE_NAMES

    demisto.debug(f"Command being called is {command}.")

    try:
        async with AsyncClient(
            base_url=base_url,
            client_id=client_id,
            client_secret=client_secret,
            verify=verify,
            proxy=proxy,
        ) as async_client:
            if command == "test-module":
                return_results(await test_module(async_client, service_names=service_names))

            elif command == "genesys-cloud-get-events":
                should_push_events = argToBoolean(args.pop("should_push_events", False))
                events, command_results = await get_events_command(async_client, args)
                return_results(command_results)
                if should_push_events:
                    send_events_to_xsiam(events, vendor=VENDOR, product=PRODUCT)

            elif command == "fetch-events":
                last_run = demisto.getLastRun()
                next_run, events = await fetch_events_command(
                    async_client,
                    last_run=last_run,
                    max_fetch=max_fetch,
                    service_names=service_names,
                )
                send_events_to_xsiam(events, vendor=VENDOR, product=PRODUCT)
                demisto.setLastRun(next_run)

            else:
                raise NotImplementedError(f"{command} command is not implemented.")

    except Exception as e:
        return_error(f"Failed to execute {command} command.\nError:\n{str(e)}")


""" ENTRY POINT """

if __name__ in ("__main__", "__builtin__", "builtins"):
    asyncio.run(main())