"""RestSession class for creating connections to the Webex APIs. Copyright (c) 2016-2024 Cisco and/or its affiliates. Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ import json import logging import platform import sys import time import urllib import urllib.parse import warnings import requests from ._metadata import __title__, __version__ from .config import DEFAULT_SINGLE_REQUEST_TIMEOUT, DEFAULT_WAIT_ON_RATE_LIMIT from .exceptions import MalformedResponse, RateLimitError, RateLimitWarning from .response_codes import EXPECTED_RESPONSE_CODE from .utils import ( check_response_code, check_type, extract_and_parse_json, validate_base_url, ) logger = logging.getLogger(__name__) # Helper Functions def _fix_next_url(next_url, params): """Remove max=null parameter from URL. Patch for Webex Defect: "next" URL returned in the Link headers of the responses contain an errant "max=null" parameter, which causes the next request (to this URL) to fail if the URL is requested as-is. This patch parses the next_url to remove the max=null parameter. Args: next_url(str): The "next" URL to be parsed and cleaned. Returns: str: The clean URL to be used for the "next" request. Raises: AssertionError: If the parameter types are incorrect. ValueError: If "next_url" does not contain a valid API endpoint URL (scheme, netloc and path). """ next_url = str(next_url) parsed_url = urllib.parse.urlparse(next_url) if not parsed_url.scheme or not parsed_url.netloc or not parsed_url.path: raise ValueError( "`next_url` must be a valid API endpoint URL, minimally " "containing a scheme, netloc and path." ) if parsed_url.query: query_list = parsed_url.query.split("&") if "max=null" in query_list: query_list.remove("max=null") warnings.warn( "`max=null` still present in next-URL returned " "from Webex", RuntimeWarning, stacklevel=1, ) if params: for k, v in params.items(): if not any(p.startswith("{}=".format(k)) for p in query_list): query_list.append("{}={}".format(k, v)) new_query = "&".join(query_list) parsed_url = list(parsed_url) parsed_url[4] = new_query return urllib.parse.urlunparse(parsed_url) def user_agent(be_geo_id=None, caller=None): """Build a User-Agent HTTP header string.""" product = __title__ version = __version__ # Add platform data to comment portion of the User-Agent header. # Inspired by PIP"s User-Agent header; serialize the data in JSON format. # https://github.com/pypa/pip/blob/master/src/pip/_internal/network data = dict() # Python implementation data["implementation"] = { "name": platform.python_implementation(), } # Implementation version if data["implementation"]["name"] == "CPython": data["implementation"]["version"] = platform.python_version() elif data["implementation"]["name"] == "PyPy": if sys.pypy_version_info.releaselevel == "final": pypy_version_info = sys.pypy_version_info[:3] else: pypy_version_info = sys.pypy_version_info data["implementation"]["version"] = ".".join( [str(x) for x in pypy_version_info] ) elif data["implementation"]["name"] == "Jython": data["implementation"]["version"] = platform.python_version() elif data["implementation"]["name"] == "IronPython": data["implementation"]["version"] = platform.python_version() # Platform information if sys.platform.startswith("darwin") and platform.mac_ver()[0]: dist = {"name": "macOS", "version": platform.mac_ver()[0]} data["distro"] = dist if platform.system(): data.setdefault("system", {})["name"] = platform.system() if platform.release(): data.setdefault("system", {})["release"] = platform.release() if platform.machine(): data["cpu"] = platform.machine() data["organization"] = {} # Add self-identified organization information to the User-Agent Header. if be_geo_id: data["organization"]["be_geo_id"] = be_geo_id if caller: data["organization"]["caller"] = caller # Create the User-Agent string user_agent_string = "{product}/{version} {comment}".format( product=product, version=version, comment=json.dumps(data), ) logger.info("User-Agent: " + user_agent_string) return user_agent_string # Main module interface class RestSession(object): """RESTful HTTP session class for making calls to the Webex APIs.""" def __init__( self, access_token, base_url, single_request_timeout=DEFAULT_SINGLE_REQUEST_TIMEOUT, wait_on_rate_limit=DEFAULT_WAIT_ON_RATE_LIMIT, proxies=None, be_geo_id=None, caller=None, disable_ssl_verify=False, ): """Initialize a new RestSession object. Args: access_token(str): The Webex access token to be used for this session. base_url(str): The base URL that will be suffixed onto API endpoint relative URLs to produce a callable absolute URL. single_request_timeout(int): The timeout (seconds) for a single HTTP REST API request. wait_on_rate_limit(bool): Enable or disable automatic rate-limit handling. proxies(dict): Dictionary of proxies passed on to the requests session. be_geo_id(str): Optional partner identifier for API usage tracking. Defaults to checking for a BE_GEO_ID environment variable. caller(str): Optional identifier for API usage tracking. Defaults to checking for a WEBEX_PYTHON_SDK_CALLER environment variable. disable_ssl_verify(bool): Optional boolean flag to disable ssl verification. Defaults to False. If set to true, the requests session won't verify ssl certs anymore. Raises: TypeError: If the parameter types are incorrect. """ check_type(access_token, str) check_type(base_url, str) check_type(single_request_timeout, int, optional=True) check_type(wait_on_rate_limit, bool) check_type(proxies, dict, optional=True) check_type(disable_ssl_verify, bool, optional=True) super(RestSession, self).__init__() # Initialize attributes and properties self._base_url = str(validate_base_url(base_url)) self._access_token = str(access_token) self._single_request_timeout = single_request_timeout self._wait_on_rate_limit = wait_on_rate_limit # Initialize a new session self._req_session = requests.session() # Disable ssl cert verification if chosen by user if disable_ssl_verify: self._req_session.verify = False if proxies is not None: self._req_session.proxies.update(proxies) # Update the HTTP headers for the session self.update_headers( { "Authorization": "Bearer " + access_token, "Content-type": "application/json;charset=utf-8", "User-Agent": user_agent(be_geo_id=be_geo_id, caller=caller), } ) @property def base_url(self): """The base URL for the API endpoints.""" return self._base_url @property def access_token(self): """The Webex access token used for this session.""" return self._access_token @property def single_request_timeout(self): """The timeout (seconds) for a single HTTP REST API request.""" return self._single_request_timeout @single_request_timeout.setter def single_request_timeout(self, value): """The timeout (seconds) for a single HTTP REST API request.""" check_type(value, int, optional=True) if value is not None and value <= 0: raise ValueError("single_request_timeout must be positive integer") self._single_request_timeout = value @property def wait_on_rate_limit(self): """Automatic rate-limit handling. This setting enables or disables automatic rate-limit handling. When enabled, rate-limited requests will be automatically be retried after waiting `Retry-After` seconds (provided by Webex in the rate-limit response header). """ return self._wait_on_rate_limit @wait_on_rate_limit.setter def wait_on_rate_limit(self, value): """Enable or disable automatic rate-limit handling.""" check_type(value, bool) self._wait_on_rate_limit = value @property def headers(self): """The HTTP headers used for requests in this session.""" return self._req_session.headers.copy() def update_headers(self, headers): """Update the HTTP headers used for requests in this session. Note: Updates provided by the dictionary passed as the `headers` parameter to this method are merged into the session headers by adding new key-value pairs and/or updating the values of existing keys. The session headers are not replaced by the provided dictionary. Args: headers(dict): Updates to the current session headers. """ check_type(headers, dict) self._req_session.headers.update(headers) def abs_url(self, url): """Given a relative or absolute URL; return an absolute URL. Args: url(str): A relative or absolute URL. Returns: str: An absolute URL. """ parsed_url = urllib.parse.urlparse(url) if not parsed_url.scheme and not parsed_url.netloc: # url is a relative URL; combine with base_url return urllib.parse.urljoin(str(self.base_url), str(url)) else: # url is already an absolute URL; return as is return url def request(self, method, url, erc, **kwargs): """Abstract base method for making requests to the Webex APIs. This base method: * Expands the API endpoint URL to an absolute URL * Makes the actual HTTP request to the API endpoint * Provides support for Webex rate-limiting * Inspects response codes and raises exceptions as appropriate Args: method(str): The request-method type ("GET", "POST", etc.). url(str): The URL of the API endpoint to be called. erc(int): The expected response code that should be returned by the Webex API endpoint to indicate success. **kwargs: Passed on to the requests package. Raises: ApiError: If anything other than the expected response code is returned by the Webex API endpoint. """ # Ensure the url is an absolute URL abs_url = self.abs_url(url) # Update request kwargs with session defaults kwargs.setdefault("timeout", self.single_request_timeout) while True: # Make the HTTP request to the API endpoint response = self._req_session.request(method, abs_url, **kwargs) try: # Check the response code for error conditions check_response_code(response, erc) except RateLimitError as e: # Catch rate-limit errors # Wait and retry if automatic rate-limit handling is enabled if self.wait_on_rate_limit: warnings.warn(RateLimitWarning(response), stacklevel=1) time.sleep(e.retry_after) continue else: # Re-raise the RateLimitError raise else: return response def get(self, url, params=None, **kwargs): """Sends a GET request. Args: url(str): The URL of the API endpoint. params(dict): The parameters for the HTTP GET request. **kwargs: erc(int): The expected (success) response code for the request. others: Passed on to the requests package. Raises: ApiError: If anything other than the expected response code is returned by the Webex API endpoint. """ check_type(url, str) check_type(params, dict, optional=True) # Expected response code erc = kwargs.pop("erc", EXPECTED_RESPONSE_CODE["GET"]) response = self.request("GET", url, erc, params=params, **kwargs) return extract_and_parse_json(response) def get_pages(self, url, params=None, **kwargs): """Return a generator that GETs and yields pages of data. Provides native support for RFC5988 Web Linking. Args: url(str): The URL of the API endpoint. params(dict): The parameters for the HTTP GET request. **kwargs: erc(int): The expected (success) response code for the request. others: Passed on to the requests package. Raises: ApiError: If anything other than the expected response code is returned by the Webex API endpoint. """ check_type(url, str) check_type(params, dict, optional=True) # Expected response code erc = kwargs.pop("erc", EXPECTED_RESPONSE_CODE["GET"]) # First request response = self.request("GET", url, erc, params=params, **kwargs) while True: yield extract_and_parse_json(response) if response.links.get("next"): next_url = response.links.get("next").get("url") # Patch for Webex "max=null" in next URL bug + missing # params, like mentionedPeople, which can be mandatory for # bots # Testing shows that patch is no longer needed; raising a # warnning if it is still taking effect; # considering for future removal next_url = _fix_next_url(next_url, params) # Subsequent requests response = self.request("GET", next_url, erc, **kwargs) else: break def get_items(self, url, params=None, **kwargs): """Return a generator that GETs and yields individual JSON `items`. Yields individual `items` from Webex"s top-level {"items": [...]} JSON objects. Provides native support for RFC5988 Web Linking. The generator will request additional pages as needed until all items have been returned. Args: url(str): The URL of the API endpoint. params(dict): The parameters for the HTTP GET request. **kwargs: erc(int): The expected (success) response code for the request. others: Passed on to the requests package. Raises: ApiError: If anything other than the expected response code is returned by the Webex API endpoint. MalformedResponse: If the returned response does not contain a top-level dictionary with an "items" key. """ # Get generator for pages of JSON data pages = self.get_pages(url, params=params, **kwargs) for json_page in pages: assert isinstance(json_page, dict) items = json_page.get("items") if items is None: error_message = ( "'items' key not found in JSON data: " "{!r}".format( json_page ) ) raise MalformedResponse(error_message) else: for item in items: yield item def post(self, url, json=None, data=None, **kwargs): """Sends a POST request. Args: url(str): The URL of the API endpoint. json: Data to be sent in JSON format in tbe body of the request. data: Data to be sent in the body of the request. **kwargs: erc(int): The expected (success) response code for the request. others: Passed on to the requests package. Raises: ApiError: If anything other than the expected response code is returned by the Webex API endpoint. """ check_type(url, str) # Expected response code erc = kwargs.pop("erc", EXPECTED_RESPONSE_CODE["POST"]) response = self.request( "POST", url, erc, json=json, data=data, **kwargs ) return extract_and_parse_json(response) def put(self, url, json=None, data=None, **kwargs): """Sends a PUT request. Args: url(str): The URL of the API endpoint. json: Data to be sent in JSON format in tbe body of the request. data: Data to be sent in the body of the request. **kwargs: erc(int): The expected (success) response code for the request. others: Passed on to the requests package. Raises: ApiError: If anything other than the expected response code is returned by the Webex API endpoint. """ check_type(url, str) # Expected response code erc = kwargs.pop("erc", EXPECTED_RESPONSE_CODE["PUT"]) response = self.request( "PUT", url, erc, json=json, data=data, **kwargs ) return extract_and_parse_json(response) def delete(self, url, **kwargs): """Sends a DELETE request. Args: url(str): The URL of the API endpoint. **kwargs: erc(int): The expected (success) response code for the request. others: Passed on to the requests package. Raises: ApiError: If anything other than the expected response code is returned by the Webex API endpoint. """ check_type(url, str) # Expected response code erc = kwargs.pop("erc", EXPECTED_RESPONSE_CODE["DELETE"]) self.request("DELETE", url, erc, **kwargs)