import json import re import typing from urllib.parse import urlparse import phonenumbers from django.conf import settings from phonenumbers import NumberParseException from telegram import Bot from twilio.base.exceptions import TwilioException from twilio.rest import Client from common.api_helpers.utils import create_engine_url class LiveSettingProxy: def __dir__(self): from apps.base.models import LiveSetting return LiveSetting.AVAILABLE_NAMES def __getattr__(self, item): from apps.base.models import LiveSetting value = LiveSetting.get_setting(item) return value def __setattr__(self, key, value): from apps.base.models import LiveSetting LiveSetting.objects.update_or_create(name=key, defaults={"value": value}) live_settings = LiveSettingProxy() class LiveSettingValidator: EMPTY_VALID_NAMES = ( "TWILIO_AUTH_TOKEN", "TWILIO_API_KEY_SID", "TWILIO_API_KEY_SECRET", ) EMAIL_SSL_TLS_ERROR_MSG = "Cannot set Email (SMTP) to use SSL and TLS at the same time" def __init__(self, live_setting): self.live_setting = live_setting def get_error(self): check_fn_name = f"_check_{self.live_setting.name.lower()}" if self.live_setting.value in (None, "") and self.live_setting.name not in self.EMPTY_VALID_NAMES: return "Empty" # skip validation if there's no handler for it if not hasattr(self, check_fn_name): return None check_fn = getattr(self, check_fn_name) return check_fn(self.live_setting.value) @classmethod def _check_twilio_api_key_sid(cls, twilio_api_key_sid): if live_settings.TWILIO_AUTH_TOKEN: return try: Client( twilio_api_key_sid, live_settings.TWILIO_API_KEY_SECRET, live_settings.TWILIO_ACCOUNT_SID ).api.applications.list(limit=1) except Exception as e: return cls._prettify_twilio_error(e) @classmethod def _check_twilio_api_key_secret(cls, twilio_api_key_secret): if live_settings.TWILIO_AUTH_TOKEN: return try: Client( live_settings.TWILIO_API_KEY_SID, twilio_api_key_secret, live_settings.TWILIO_ACCOUNT_SID ).api.applications.list(limit=1) except Exception as e: return cls._prettify_twilio_error(e) @classmethod def _check_twilio_account_sid(cls, twilio_account_sid): try: if live_settings.TWILIO_API_KEY_SID and live_settings.TWILIO_API_KEY_SECRET: Client( live_settings.TWILIO_API_KEY_SID, live_settings.TWILIO_API_KEY_SECRET, twilio_account_sid ).api.applications.list(limit=1) else: Client(twilio_account_sid, live_settings.TWILIO_AUTH_TOKEN).api.applications.list(limit=1) except Exception as e: return cls._prettify_twilio_error(e) @classmethod def _check_twilio_auth_token(cls, twilio_auth_token): if live_settings.TWILIO_API_KEY_SID and live_settings.TWILIO_API_KEY_SECRET: return try: Client(live_settings.TWILIO_ACCOUNT_SID, twilio_auth_token).api.applications.list(limit=1) except Exception as e: return cls._prettify_twilio_error(e) @classmethod def _check_twilio_verify_service_sid(cls, twilio_verify_service_sid): try: if live_settings.TWILIO_API_KEY_SID and live_settings.TWILIO_API_KEY_SECRET: twilio_client = Client( live_settings.TWILIO_API_KEY_SID, live_settings.TWILIO_API_KEY_SECRET, live_settings.TWILIO_ACCOUNT_SID, ) else: twilio_client = Client(live_settings.TWILIO_ACCOUNT_SID, live_settings.TWILIO_AUTH_TOKEN) twilio_client.verify.services(twilio_verify_service_sid).rate_limits.list(limit=1) except Exception as e: return cls._prettify_twilio_error(e) @classmethod def _check_twilio_number(cls, twilio_number): if not cls._is_phone_number_valid(twilio_number): return "Please specify a valid phone number in the following format: +XXXXXXXXXXX" @classmethod def _check_slack_install_return_redirect_host(cls, slack_install_return_redirect_host): scheme = urlparse(slack_install_return_redirect_host).scheme if scheme != "https": return "Must use https" @classmethod def _check_telegram_token(cls, telegram_token): try: bot = Bot(telegram_token) bot.get_me() except Exception as e: return f"Telegram error: {str(e)}" @classmethod def _check_telegram_webhook_host(cls, telegram_webhook_host): if settings.FEATURE_TELEGRAM_LONG_POLLING_ENABLED: return try: # avoid circular import from apps.telegram.client import TelegramClient url = create_engine_url("/telegram/", override_base=telegram_webhook_host) TelegramClient().register_webhook(url) except Exception as e: return f"Telegram error: {str(e)}" @classmethod def _check_grafana_cloud_oncall_token(cls, grafana_oncall_token): from apps.oss_installation.models import CloudConnector _, err = CloudConnector.sync_with_cloud(grafana_oncall_token) return err @classmethod def _check_email_use_tls(cls, email_use_tls: bool) -> typing.Optional[str]: return cls.EMAIL_SSL_TLS_ERROR_MSG if live_settings.EMAIL_USE_SSL is True and email_use_tls is True else None @classmethod def _check_email_use_ssl(cls, email_use_ssl: bool) -> typing.Optional[str]: return cls.EMAIL_SSL_TLS_ERROR_MSG if live_settings.EMAIL_USE_TLS is True and email_use_ssl is True else None @staticmethod def _is_email_valid(email): return re.match(r"^[^@]+@[^@]+\.[^@]+$", email) @staticmethod def _is_phone_number_valid(phone_number): try: ph_num = phonenumbers.parse(phone_number) return phonenumbers.is_valid_number(ph_num) except NumberParseException: return False @staticmethod def _prettify_twilio_error(exc): if isinstance(exc, TwilioException): if len(exc.args) > 1: response_content = exc.args[1].content content = json.loads(response_content) error_code = content["code"] more_info = content["more_info"] return f"Twilio error: code {error_code}. Learn more: {more_info}" else: return f"Twilio error: {exc.args[0]}" else: return f"Twilio error: {str(exc)}"