configuration.py

# coding: utf-8

"""
Copyright 2016 SmartBear Software

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.

   ref: https://github.com/swagger-api/swagger-codegen
"""

from __future__ import absolute_import
import base64
import urllib3

try:
    import httplib
except ImportError:
    # for python3
    import http.client as httplib

import sys
from os.path import expanduser
import configparser
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, FileModifiedEvent
import os
import time
import threading
import hashlib
import json
import io
import ssl
import certifi

from .logger import Logger, LogFormat, LogLevel
from .gateway_configuration import GatewayConfiguration

def singleton(cls, *args, **kw):
    instances = {}

    def _singleton():
        if cls not in instances:
            instances[cls] = cls(*args, **kw)
        return instances[cls]
    return _singleton


@singleton
class Configuration(object):
    """
    NOTE: This class is auto generated by the swagger code generator program.
    Ref: https://github.com/swagger-api/swagger-codegen
    Do not edit the class manually.
    """

    def __init__(self):
        """
        Constructor
        """
        # Default Base url
        self.host = "https://api.mypurecloud.com"
        # Default api client
        self.api_client = None
        # Temp file folder for downloading files
        self.temp_folder_path = None

        # Authentication Settings
        # dict to store API key(s)
        self.api_key = {}
        # dict to store API prefix (e.g. Bearer)
        self.api_key_prefix = {}
        # Username for HTTP basic authentication
        self.username = ""
        # Password for HTTP basic authentication
        self.password = ""

        # access token for OAuth
        self.access_token = ""
        # used to determine if access token should be refresh transparently when using Code Authorization
        self.should_refresh_access_token = True
        # maximum amount of time other threads will wait for a thread to request a new access token when it expires
        self.refresh_token_wait_time = 10


        # SSL/TLS verification
        # Set this to false to skip verifying SSL certificate when calling API from https server.
        self.verify_ssl = True
        # Set this to customize the certificate file to verify the peer.
        self.ssl_ca_cert = None
        # client certificate file
        self.cert_file = None
        # client key file
        self.key_file = None
        #ssl context
        self.ssl_context = None

        # proxy
        self.proxy = None

        # proxy username
        self.proxy_username = None

        # proxy password
        self.proxy_password = None

        # gateway configuration
        self.gateway_configuration = None

        # Logging Settings
        self.logger = Logger()

        # Private config file variables
        # path to the config file
        # default to ~/.genesyscloudpython/config
        self.config_file_path = os.path.join(expanduser("~"), ".genesyscloudpython", "config")
        # private directory observer instance
        self._observer = None

        # flag to control running of _config_updater thread
        self.live_reload_config = True
 
        # update config from config file if possible
        self._update_config_from_file()

        # if live_reload_config set, start the config_updater thread
        if self.live_reload_config:
            run_observer = threading.Thread(target=self._run_observer)
            run_observer.setDaemon(True)
            run_observer.start()

            self._config_updater()

    @property
    def config_file_path(self):
        return self.__config_file_path

    @config_file_path.setter
    def config_file_path(self, value):
        self.__config_file_path = value
        self._update_config_from_file()
        if not hasattr(self, "_observer"):
            return
        if self.live_reload_config:
            self._config_updater()

    def create_mtls_or_ssl_context(self):
        if self.ssl_context is None:  # set_mtls_contents() or set_mtls_certificates() were not called.
            ssl_context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)

            if self.ssl_ca_cert:
                ssl_context.load_verify_locations(cafile=self.ssl_ca_cert)
            else:
                # If no CA certs are provided, use certifi's bundle.
                ssl_context.load_verify_locations(cafile=certifi.where())

            if self.cert_file and self.key_file:
                # mTLS configuration
                ssl_context.load_cert_chain(certfile=self.cert_file, keyfile=self.key_file)

            self.ssl_context = ssl_context

    def get_api_key_with_prefix(self, identifier):
        """
        Gets API key (with prefix if set).

        :param identifier: The identifier of apiKey.
        :return: The token for api key authentication.
        """
        if self.api_key.get(identifier) and self.api_key_prefix.get(identifier):
            return self.api_key_prefix[identifier] + ' ' + self.api_key[identifier]
        elif self.api_key.get(identifier):
            return self.api_key[identifier]

    def get_basic_auth_token(self):
        """
        Gets HTTP basic authentication header (string).

        :return: The token for basic HTTP authentication.
        """
        return urllib3.util.make_headers(basic_auth=self.username + ':' + self.password)\
                           .get('authorization')

    def auth_settings(self, access_token=None):
        """
        Gets Auth Settings dict for api client.

        :return: The Auth Settings information dict.
        """

        return {
            'Guest Chat JWT':
                {
                    'type': 'api_key',
                    'in': 'header',
                    'key': 'Authorization',
                    'value': self.get_api_key_with_prefix('Authorization')
                },

            'PureCloud OAuth':
                {
                    'type': 'oauth2',
                    'in': 'header',
                    'key': 'Authorization',
                    'value': 'Bearer ' + self.access_token if access_token is None or access_token == "" else 'Bearer ' + access_token
                },

        }

    def to_debug_report(self):
        """
        Gets the essential information for debugging.

        :return: The report for debugging.
        """
        return "Python SDK Debug Report:\n"\
               "OS: {env}\n"\
               "Python Version: {pyversion}\n"\
               "Version of the API: v2\n"\
               "SDK Package Version: 227.0.0".\
               format(env=sys.platform, pyversion=sys.version)

    def _update_config_from_file(self):
        try:
            config = configparser.ConfigParser()
            # try to parse as INI format
            try:
                # if it doesn't exist, this function will return []
                if config.read(self.config_file_path) == []:
                    return
            except configparser.MissingSectionHeaderError as e:
                # this exception means it's possibly JSON
                try:
                    with open(self.config_file_path, "r") as read_file:
                        config = json.load(read_file)
                except Exception:
                    return
            # logging
            log_level = _get_config_string(config, "logging", "log_level")
            if log_level is not None:
                self.logger.log_level = LogLevel.from_string(log_level)

            log_format = _get_config_string(config, "logging", "log_format")
            if log_format is not None:
                self.logger.log_format = LogFormat.from_string(log_format)
            
            log_to_console = _get_config_bool(config, "logging", "log_to_console")
            if log_to_console is not None:
                self.logger.log_to_console = log_to_console

            log_file_path = _get_config_string(config, "logging", "log_file_path")
            if log_file_path is not None:
                self.logger.log_file_path = log_file_path

            log_response_body = _get_config_bool(config, "logging", "log_response_body")
            if log_response_body is not None:
                self.logger.log_response_body = log_response_body

            log_request_body = _get_config_bool(config, "logging", "log_request_body")
            if log_request_body is not None:
                self.logger.log_request_body = log_request_body
            
            # general
            host = _get_config_string(config, "general", "host")
            if host is not None:
                self.host = host

            live_reload_config = _get_config_bool(config, "general", "live_reload_config")
            if live_reload_config is not None:
                self.live_reload_config = live_reload_config

            # reauthentication
            refresh_access_token = _get_config_bool(config, "reauthentication", "refresh_access_token")
            if refresh_access_token is not None:
                self.should_refresh_access_token = refresh_access_token
            
            refresh_token_wait_max = _get_config_int(config, "reauthentication", "refresh_token_wait_max")
            if refresh_token_wait_max is not None:
                self.refresh_token_wait_time = refresh_token_wait_max
            
            # gateway configuration
            gateway_host = _get_config_string(config, "gateway", "host")
            if not gateway_host:
                self.gateway_configuration = None
            else:
                self.gateway_configuration = GatewayConfiguration(host = gateway_host)
                gateway_protocol = _get_config_string(config, "gateway", "protocol")
                if gateway_protocol is not None:
                    self.gateway_configuration.protocol = gateway_protocol;
                gateway_port = _get_config_int(config, "gateway", "port")
                if gateway_port is not None:
                    self.gateway_configuration.port = gateway_port;
                gateway_path_params_login = _get_config_string(config, "gateway", "path_params_login")
                if gateway_path_params_login is not None:
                    self.gateway_configuration.path_params_login = gateway_path_params_login;
                gateway_path_params_api = _get_config_string(config, "gateway", "path_params_api")
                if gateway_path_params_api is not None:
                    self.gateway_configuration.path_params_api = gateway_path_params_api;
                gateway_username = _get_config_string(config, "gateway", "username")
                if gateway_username is not None:
                    self.gateway_configuration.username = gateway_username;
                gateway_password = _get_config_string(config, "gateway", "password")
                if gateway_password is not None:
                    self.gateway_configuration.password = gateway_password;
        except Exception:
            return

    def _run_observer(self):
        self._observer = Observer()
        self._observer.start()
        try:
            while True:
                time.sleep(1)
        finally:
            self._observer.stop()
            self._observer.join()

    def _config_updater(self):
        if self._observer is not None:
            self._observer.unschedule_all()
        event_handler = ConfigFileEventHandler(self)

        # watch the parent directory of the config file
        watched_directory = os.path.dirname(self.config_file_path)
        # go up the directory tree if the parent directory doesn't yet exist
        while not os.path.exists(watched_directory):
            watched_directory = os.path.dirname(watched_directory)
            if watched_directory == "":
                return

        while True:
            try:
                self._observer.schedule(event_handler, watched_directory, recursive=True)
                break
            except FileNotFoundError:
                watched_directory = os.path.dirname(watched_directory)
                if watched_directory == "":
                    return
            except Exception as e:
                return

def _get_config_string(config, section, key):
        try:
            return str(config[section][key]).strip()
        except:
            return None

def _get_config_bool(config, section, key):
        try:
            if type(config) == configparser.ConfigParser:
                return config.getboolean(section, key)
            else:
                return config[section][key]
        except:
            return None

def _get_config_int(config, section, key):
        try:
            if type(config) == configparser.ConfigParser:
                return config.getint(section, key)
            else:
                return config[section][key]
        except:
            return None

class ConfigFileEventHandler(FileSystemEventHandler):
    def __init__(self, configuration):
        super(FileSystemEventHandler, self).__init__()
        self.configuration = configuration
        self.config_file_path = self.configuration.config_file_path

    def on_modified(self, event):
        # only respond if the config file has been modified
        if type(event) == FileModifiedEvent and event.src_path == self.config_file_path:
            if self.configuration.live_reload_config:
                self.configuration._update_config_from_file()