ZoomApiModule

Common Zoom code that provides generic infrastructure, and will be appended to each Zoom integration when it is deployed.

Type

python

Pack

ApiModules

Source

from datetime import timedelta

import dateparser
import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401

""" CONSTANTS """

OAUTH_TOKEN_GENERATOR_URL = "https://zoom.us/oauth/token"
OAUTH_OGV_TOKEN_GENERATOR_URL = "https://zoomgov.com/oauth/token"
# The token's time to live is 1 hour,
# two minutes were subtract for extra safety.
TOKEN_LIFE_TIME = timedelta(minutes=58)
# the lifetime for an JWT token is 90 minutes == 5400 seconds
# 400 seconds were subtract for extra safety.
JWT_LIFETIME = 5000
# maximun records that the api can return in one request
MAX_RECORDS_PER_PAGE = 300

# ERRORS
INVALID_CREDENTIALS = "Invalid credentials. Please verify that your credentials are valid."
INVALID_API_SECRET = "Invalid API Secret. Please verify that your API Secret is valid."
INVALID_ID_OR_SECRET = "Invalid Client ID or Client Secret. Please verify that your ID and Secret is valid."
INVALID_TOKEN = "Invalid Authorization token. Please verify that your Bot ID and Bot Secret is valid."
INVALID_BOT_ID = "No Chatbot can be found with the given robot_jid value. Please verify that your Bot JID is correct"
"""CLIENT CLASS"""


class Zoom_Client(BaseClient):
    """A client class that implements logic to authenticate with Zoom application."""

    def __init__(
        self,
        base_url: str,
        api_key: str | None = None,
        api_secret: str | None = None,
        account_id: str | None = None,
        client_id: str | None = None,
        client_secret: str | None = None,
        bot_client_id: str | None = None,
        bot_client_secret: str | None = None,
        verify=True,
        proxy=False,
        bot_jid: str | None = None,
    ):
        super().__init__(base_url, verify, proxy)
        self.api_key = api_key
        self.api_secret = api_secret
        self.account_id = account_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.bot_client_id = bot_client_id
        self.bot_client_secret = bot_client_secret
        self.bot_jid = bot_jid
        try:
            self.access_token, self.bot_access_token = self.get_oauth_token()
        except Exception as e:
            demisto.info(f"Cannot get access token. Error: {e}")
            self.access_token = None
            self.bot_access_token = None

    def generate_oauth_token(self):
        """
        Generate an OAuth Access token using the app credentials (AKA: client id and client secret) and the account id

        :return: valid token
        """
        full_url = OAUTH_OGV_TOKEN_GENERATOR_URL if "gov" in self._base_url else OAUTH_TOKEN_GENERATOR_URL
        token_res = self._http_request(
            method="POST",
            full_url=full_url,
            params={"account_id": self.account_id, "grant_type": "account_credentials"},
            auth=(self.client_id, self.client_secret),
        )
        return token_res.get("access_token")

    def generate_oauth_client_token(self):
        """
        Generate an OAuth Access token using the app credentials (AKA: client id and client secret) and the account id

        :return: valid token
        """
        full_url = OAUTH_OGV_TOKEN_GENERATOR_URL if "gov" in self._base_url else OAUTH_TOKEN_GENERATOR_URL
        token_res = self._http_request(
            method="POST",
            full_url=full_url,
            params={"account_id": self.account_id, "grant_type": "client_credentials"},
            auth=(self.bot_client_id, self.bot_client_secret),
        )
        return token_res.get("access_token")

    def get_oauth_token(self, force_gen_new_token=False):
        """
        Retrieves the token from the server if it's expired and updates the global HEADERS to include it

        :param force_gen_new_token: If set to True will generate a new token regardless of time passed

        :rtype: ``str``
        :return: Token
        """
        now = datetime.now()
        ctx = get_integration_context()
        client_oauth_token = None
        oauth_token = None

        if not ctx or not ctx.get("token_info").get("generation_time", force_gen_new_token):
            # new token is needed
            if self.client_id and self.client_secret:
                oauth_token = self.generate_oauth_token()
            if self.bot_client_id and self.bot_client_secret:
                client_oauth_token = self.generate_oauth_client_token()
            ctx = {}
        else:
            if generation_time := dateparser.parse(ctx.get("token_info").get("generation_time")):
                time_passed = now - generation_time
            else:
                time_passed = TOKEN_LIFE_TIME
            if time_passed < TOKEN_LIFE_TIME:
                # token hasn't expired
                return ctx.get("token_info", {}).get("oauth_token"), ctx.get("token_info", {}).get("client_oauth_token")
            else:
                # token expired
                # new token is needed
                if self.client_id and self.client_secret:
                    oauth_token = self.generate_oauth_token()
                if self.bot_client_id and self.bot_client_secret:
                    client_oauth_token = self.generate_oauth_client_token()

        ctx.update(
            {
                "token_info": {
                    "oauth_token": oauth_token,
                    "client_oauth_token": client_oauth_token,
                    "generation_time": now.strftime("%Y-%m-%dT%H:%M:%S"),
                }
            }
        )
        set_integration_context(ctx)
        return oauth_token, client_oauth_token

    def error_handled_http_request(
        self,
        method,
        url_suffix="",
        full_url=None,
        headers=None,
        auth=None,
        json_data=None,
        params=None,
        files=None,
        data=None,
        return_empty_response: bool = False,
        resp_type: str = "json",
        stream: bool = False,
    ):
        # all future functions should call this function instead of the original _http_request.
        # This is needed because the OAuth token may not behave consistently,
        # First the func will make an http request with a token,
        # and if it turns out to be invalid, the func will retry again with a new token.
        try:
            return super()._http_request(
                method=method,
                url_suffix=url_suffix,
                full_url=full_url,
                headers=headers,
                auth=auth,
                json_data=json_data,
                params=params,
                files=files,
                data=data,
                return_empty_response=return_empty_response,
                resp_type=resp_type,
                stream=stream,
            )
        except DemistoException as e:
            if any(
                message in e.message
                for message in ["Invalid access token", "Access token is expired.", "Invalid authorization token"]
            ):
                if url_suffix == "/im/chat/messages":
                    demisto.debug("generate new bot client token")
                    self.bot_access_token = self.generate_oauth_client_token()
                    headers = {"authorization": f"Bearer {self.bot_access_token}"}
                else:
                    self.access_token = self.generate_oauth_token()
                    headers = {"authorization": f"Bearer {self.access_token}"}
                return super()._http_request(
                    method=method,
                    url_suffix=url_suffix,
                    full_url=full_url,
                    headers=headers,
                    auth=auth,
                    json_data=json_data,
                    params=params,
                    files=files,
                    data=data,
                    return_empty_response=return_empty_response,
                    resp_type=resp_type,
                    stream=stream,
                )
            else:
                raise DemistoException(e.message, url_suffix)


""" HELPER FUNCTIONS """

README

def main():
    ...


from ZoomApiModule import *  # noqa: E402

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