import json import logging import requests from django.conf import settings from django.core.cache import cache from license_info.exeptions import LicenseException LICENSE_CLIENT_URL = getattr(settings, 'LICENSE_CLIENT_URL', '') LICENSE_CACHE_TIMEOUT = getattr(settings, 'LICENSE_CACHE_TIMEOUT', 60 * 60) LICENSE_KEY = "license_key" LICENSE_INFO_URL = f"{LICENSE_CLIENT_URL}/license/" LICENSE_TOKEN_URL = f"{LICENSE_CLIENT_URL}/token/" LICENSE_ACTIVATE_URL = f"{LICENSE_CLIENT_URL}/activate/" LICENSE_AUTO_ACTIVATE_URL = f"{LICENSE_CLIENT_URL}/auto/" _log = logging.getLogger() def check_response(response): """Check that license client's response has no error @param response: is a requests.response object @raise RuntimeError if response isn't good @return parsed response""" if response.status_code != 200: _log.error(f"Can't get license info {response}") raise LicenseException("Bad license client response code") try: data = response.json() except json.JSONDecodeError: _log.error(f"Can't parse json response: {response.content}") raise LicenseException("Can't parse client's json") # Need to check for error if data["status"] != "ok": _log.error(f"Got error from client: {data['reason']}") raise LicenseException("Bad client response") return data["response"] def get_license_info(): """Get current license info This function will use cached value most of the time @raise RuntimeError if something go wrong @return license info""" license_data = cache.get(LICENSE_KEY) if isinstance(license_data, dict): return license_data # Need to query for license info from client response = requests.get(LICENSE_INFO_URL) license_data = check_response(response) try: license_data['options']['event_sources'] = int(license_data['options']['event_sources']) except KeyError: if 'options' not in license_data: license_data['options'] = {} if 'event_sources' not in license_data['options']: license_data['options']['event_sources'] = 0 except ValueError: license_data['options']['event_sources'] = 0 # Save license to cache cache.set(LICENSE_KEY, license_data, LICENSE_CACHE_TIMEOUT) _log.debug("License information updated") return license_data def get_activation_token(serial): """Get activation token @param serial: application serial number @raise RuntimeError if something goes wrong @return activation token as string""" response = requests.get(LICENSE_TOKEN_URL, params={"serial": serial}) return check_response(response) def activate_license(license_json): """Activate license from license server @param license_json: License that we got from license server @raise RuntimeError if something goes wrong""" response = requests.post(LICENSE_ACTIVATE_URL, json=license_json) data = check_response(response) cache.set(LICENSE_KEY, data, LICENSE_CACHE_TIMEOUT) _log.info("License activated") def activate_license_auto(serial): """Activate license via activations server @param serial: application serial number @raise RuntimeError if something goes wrong""" response = requests.get(LICENSE_AUTO_ACTIVATE_URL, params={"serial": serial}) data = check_response(response) # Save license to cache cache.set(LICENSE_KEY, data, LICENSE_CACHE_TIMEOUT) _log.info("License activated via license server") def check_features(features): """ Check if license has requested features @param features: List of strings with features names @return: True if license has ALL features and False if not """ try: info = get_license_info() if isinstance(features, (list, tuple)): for cur in features: if cur not in info["features"]: return False return True elif isinstance(features, str): return features in info["features"] else: return False except: return False