[ceph.git] / ceph / src / pybind / mgr / dashboard / settings.py

# -*- coding: utf-8 -*-
import errno
import inspect
from ast import literal_eval
from typing import Any

from mgr_module import CLICheckNonemptyFileInput

from . import mgr


class Setting:
    """
    Setting representation that allows to set a default value and a list of allowed data types.
    :param default_value: The name of the bucket.
    :param types: a list consisting of the primary/preferred type and, optionally,
    secondary/legacy types for backward compatibility.
    """

    def __init__(self, default_value: Any, types: list):
        if not isinstance(types, list):
            raise ValueError('Setting types must be a list.')
        default_value_type = type(default_value)
        if default_value_type not in types:
            raise ValueError('Default value type not allowed.')
        self.default_value = default_value
        self.types = types

    def types_as_str(self):
        return ','.join([x.__name__ for x in self.types])

    def cast(self, value):
        for type_index, setting_type in enumerate(self.types):
            try:
                if setting_type.__name__ == 'bool' and str(value).lower() == 'false':
                    return False
                elif setting_type.__name__ == 'dict':
                    return literal_eval(value)
                return setting_type(value)
            except (SyntaxError, TypeError, ValueError) as error:
                if type_index == len(self.types) - 1:
                    raise error


class Options(object):
    """
    If you need to store some configuration value please add the config option
    name as a class attribute to this class.

    Example::

        GRAFANA_API_HOST = ('localhost', str)
        GRAFANA_API_PORT = (3000, int)
    """
    ENABLE_BROWSABLE_API = Setting(True, [bool])
    REST_REQUESTS_TIMEOUT = Setting(45, [int])

    # AUTHENTICATION ATTEMPTS
    ACCOUNT_LOCKOUT_ATTEMPTS = Setting(10, [int])

    # API auditing
    AUDIT_API_ENABLED = Setting(False, [bool])
    AUDIT_API_LOG_PAYLOAD = Setting(True, [bool])

    # RGW settings
    RGW_API_ACCESS_KEY = Setting('', [dict, str])
    RGW_API_SECRET_KEY = Setting('', [dict, str])
    RGW_API_ADMIN_RESOURCE = Setting('admin', [str])
    RGW_API_SSL_VERIFY = Setting(True, [bool])

    # Ceph Issue Tracker API Access Key
    ISSUE_TRACKER_API_KEY = Setting('', [str])

    # Grafana settings
    GRAFANA_API_URL = Setting('', [str])
    GRAFANA_FRONTEND_API_URL = Setting('', [str])
    GRAFANA_API_USERNAME = Setting('admin', [str])
    GRAFANA_API_PASSWORD = Setting('admin', [str])
    GRAFANA_API_SSL_VERIFY = Setting(True, [bool])
    GRAFANA_UPDATE_DASHBOARDS = Setting(False, [bool])

    # NFS Ganesha settings
    GANESHA_CLUSTERS_RADOS_POOL_NAMESPACE = Setting('', [str])

    # Prometheus settings
    PROMETHEUS_API_HOST = Setting('', [str])
    PROMETHEUS_API_SSL_VERIFY = Setting(True, [bool])
    ALERTMANAGER_API_HOST = Setting('', [str])
    ALERTMANAGER_API_SSL_VERIFY = Setting(True, [bool])

    # iSCSI management settings
    ISCSI_API_SSL_VERIFICATION = Setting(True, [bool])

    # user management settings
    # Time span of user passwords to expire in days.
    # The default value is '0' which means that user passwords are
    # never going to expire.
    USER_PWD_EXPIRATION_SPAN = Setting(0, [int])
    # warning levels to notify the user that the password is going
    # to expire soon
    USER_PWD_EXPIRATION_WARNING_1 = Setting(10, [int])
    USER_PWD_EXPIRATION_WARNING_2 = Setting(5, [int])

    # Password policy
    PWD_POLICY_ENABLED = Setting(True, [bool])
    # Individual checks
    PWD_POLICY_CHECK_LENGTH_ENABLED = Setting(True, [bool])
    PWD_POLICY_CHECK_OLDPWD_ENABLED = Setting(True, [bool])
    PWD_POLICY_CHECK_USERNAME_ENABLED = Setting(False, [bool])
    PWD_POLICY_CHECK_EXCLUSION_LIST_ENABLED = Setting(False, [bool])
    PWD_POLICY_CHECK_COMPLEXITY_ENABLED = Setting(False, [bool])
    PWD_POLICY_CHECK_SEQUENTIAL_CHARS_ENABLED = Setting(False, [bool])
    PWD_POLICY_CHECK_REPETITIVE_CHARS_ENABLED = Setting(False, [bool])
    # Settings
    PWD_POLICY_MIN_LENGTH = Setting(8, [int])
    PWD_POLICY_MIN_COMPLEXITY = Setting(10, [int])
    PWD_POLICY_EXCLUSION_LIST = Setting(','.join(['osd', 'host', 'dashboard', 'pool',
                                                  'block', 'nfs', 'ceph', 'monitors',
                                                  'gateway', 'logs', 'crush', 'maps']),
                                        [str])

    UNSAFE_TLS_v1_2 = Setting(False, [bool])

    @staticmethod
    def has_default_value(name):
        return getattr(Settings, name, None) is None or \
            getattr(Settings, name) == getattr(Options, name).default_value


class SettingsMeta(type):
    def __getattr__(cls, attr):
        setting = getattr(Options, attr)
        return setting.cast(mgr.get_module_option(attr, setting.default_value))

    def __setattr__(cls, attr, value):
        if not attr.startswith('_') and hasattr(Options, attr):
            mgr.set_module_option(attr, str(value))
        else:
            setattr(SettingsMeta, attr, value)

    def __delattr__(cls, attr):
        if not attr.startswith('_') and hasattr(Options, attr):
            mgr.set_module_option(attr, None)


# pylint: disable=no-init
class Settings(object, metaclass=SettingsMeta):
    pass


def _options_command_map():
    def filter_attr(member):
        return not inspect.isroutine(member)

    cmd_map = {}
    for option, setting in inspect.getmembers(Options, filter_attr):
        if option.startswith('_'):
            continue
        key_get = 'dashboard get-{}'.format(option.lower().replace('_', '-'))
        key_set = 'dashboard set-{}'.format(option.lower().replace('_', '-'))
        key_reset = 'dashboard reset-{}'.format(option.lower().replace('_', '-'))
        cmd_map[key_get] = {'name': option, 'type': None}
        cmd_map[key_set] = {'name': option, 'type': setting.types_as_str()}
        cmd_map[key_reset] = {'name': option, 'type': None}
    return cmd_map


_OPTIONS_COMMAND_MAP = _options_command_map()


def options_command_list():
    """
    This function generates a list of ``get`` and ``set`` commands
    for each declared configuration option in class ``Options``.
    """
    def py2ceph(pytype):
        if pytype == str:
            return 'CephString'
        elif pytype == int:
            return 'CephInt'
        return 'CephString'

    cmd_list = []
    for cmd, opt in _OPTIONS_COMMAND_MAP.items():
        if cmd.startswith('dashboard get'):
            cmd_list.append({
                'cmd': '{}'.format(cmd),
                'desc': 'Get the {} option value'.format(opt['name']),
                'perm': 'r'
            })
        elif cmd.startswith('dashboard set'):
            cmd_entry = {
                'cmd': '{} name=value,type={}'
                       .format(cmd, py2ceph(opt['type'])),
                'desc': 'Set the {} option value'.format(opt['name']),
                'perm': 'w'
            }
            if handles_secret(cmd):
                cmd_entry['cmd'] = cmd
                cmd_entry['desc'] = '{} read from -i <file>'.format(cmd_entry['desc'])
            cmd_list.append(cmd_entry)
        elif cmd.startswith('dashboard reset'):
            desc = 'Reset the {} option to its default value'.format(
                opt['name'])
            cmd_list.append({
                'cmd': '{}'.format(cmd),
                'desc': desc,
                'perm': 'w'
            })

    return cmd_list


def options_schema_list():
    def filter_attr(member):
        return not inspect.isroutine(member)

    result = []
    for option, setting in inspect.getmembers(Options, filter_attr):
        if option.startswith('_'):
            continue
        result.append({'name': option, 'default': setting.default_value,
                       'type': setting.types_as_str()})

    return result


def handle_option_command(cmd, inbuf):
    if cmd['prefix'] not in _OPTIONS_COMMAND_MAP:
        return -errno.ENOSYS, '', "Command not found '{}'".format(cmd['prefix'])

    opt = _OPTIONS_COMMAND_MAP[cmd['prefix']]

    if cmd['prefix'].startswith('dashboard reset'):
        delattr(Settings, opt['name'])
        return 0, 'Option {} reset to default value "{}"'.format(
            opt['name'], getattr(Settings, opt['name'])), ''
    elif cmd['prefix'].startswith('dashboard get'):
        return 0, str(getattr(Settings, opt['name'])), ''
    elif cmd['prefix'].startswith('dashboard set'):
        if handles_secret(cmd['prefix']):
            value, stdout, stderr = get_secret(inbuf=inbuf)
            if stderr:
                return value, stdout, stderr
        else:
            value = cmd['value']
        setting = getattr(Options, opt['name'])
        setattr(Settings, opt['name'], setting.cast(value))
        return 0, 'Option {} updated'.format(opt['name']), ''


def handles_secret(cmd: str) -> bool:
    return bool([cmd for secret_word in ['password', 'key'] if (secret_word in cmd)])


@CLICheckNonemptyFileInput(desc='password/secret')
def get_secret(inbuf=None):
    return inbuf, None, None
Commit	Line	Data
11fdf7f2	1	# -- coding: utf-8 --
11fdf7f2 TL	2	import errno
11fdf7f2 TL	3	import inspect
f67539c2 TL	4	from ast import literal_eval
f67539c2 TL	5	from typing import Any
11fdf7f2	6
cd265ab1 TL	7	from mgr_module import CLICheckNonemptyFileInput
cd265ab1 TL	8
11fdf7f2 TL	9	from . import mgr
	10
	11
f67539c2 TL	12	class Setting:
	13	"""
	14	Setting representation that allows to set a default value and a list of allowed data types.
	15	:param default_value: The name of the bucket.
	16	:param types: a list consisting of the primary/preferred type and, optionally,
	17	secondary/legacy types for backward compatibility.
	18	"""
	19
	20	def __init__(self, default_value: Any, types: list):
	21	if not isinstance(types, list):
	22	raise ValueError('Setting types must be a list.')
	23	default_value_type = type(default_value)
	24	if default_value_type not in types:
	25	raise ValueError('Default value type not allowed.')
	26	self.default_value = default_value
	27	self.types = types
	28
	29	def types_as_str(self):
	30	return ','.join([x.__name__ for x in self.types])
	31
	32	def cast(self, value):
	33	for type_index, setting_type in enumerate(self.types):
	34	try:
	35	if setting_type.__name__ == 'bool' and str(value).lower() == 'false':
	36	return False
	37	elif setting_type.__name__ == 'dict':
	38	return literal_eval(value)
	39	return setting_type(value)
	40	except (SyntaxError, TypeError, ValueError) as error:
	41	if type_index == len(self.types) - 1:
	42	raise error
	43
	44
11fdf7f2 TL	45	class Options(object):
	46	"""
	47	If you need to store some configuration value please add the config option
	48	name as a class attribute to this class.
	49
	50	Example::
	51
	52	GRAFANA_API_HOST = ('localhost', str)
	53	GRAFANA_API_PORT = (3000, int)
	54	"""
f67539c2 TL	55	ENABLE_BROWSABLE_API = Setting(True, [bool])
f67539c2 TL	56	REST_REQUESTS_TIMEOUT = Setting(45, [int])
11fdf7f2	57
adb31ebb	58	# AUTHENTICATION ATTEMPTS
f67539c2	59	ACCOUNT_LOCKOUT_ATTEMPTS = Setting(10, [int])
adb31ebb	60
11fdf7f2	61	# API auditing
f67539c2 TL	62	AUDIT_API_ENABLED = Setting(False, [bool])
f67539c2 TL	63	AUDIT_API_LOG_PAYLOAD = Setting(True, [bool])
11fdf7f2 TL	64
11fdf7f2 TL	65	# RGW settings
f67539c2 TL	66	RGW_API_ACCESS_KEY = Setting('', [dict, str])
	67	RGW_API_SECRET_KEY = Setting('', [dict, str])
	68	RGW_API_ADMIN_RESOURCE = Setting('admin', [str])
f67539c2	69	RGW_API_SSL_VERIFY = Setting(True, [bool])
11fdf7f2	70
20effc67 TL	71	# Ceph Issue Tracker API Access Key
	72	ISSUE_TRACKER_API_KEY = Setting('', [str])
	73
11fdf7f2	74	# Grafana settings
f67539c2 TL	75	GRAFANA_API_URL = Setting('', [str])
	76	GRAFANA_FRONTEND_API_URL = Setting('', [str])
	77	GRAFANA_API_USERNAME = Setting('admin', [str])
	78	GRAFANA_API_PASSWORD = Setting('admin', [str])
	79	GRAFANA_API_SSL_VERIFY = Setting(True, [bool])
	80	GRAFANA_UPDATE_DASHBOARDS = Setting(False, [bool])
11fdf7f2 TL	81
11fdf7f2 TL	82	# NFS Ganesha settings
f67539c2	83	GANESHA_CLUSTERS_RADOS_POOL_NAMESPACE = Setting('', [str])
11fdf7f2 TL	84
11fdf7f2 TL	85	# Prometheus settings
f67539c2 TL	86	PROMETHEUS_API_HOST = Setting('', [str])
	87	PROMETHEUS_API_SSL_VERIFY = Setting(True, [bool])
	88	ALERTMANAGER_API_HOST = Setting('', [str])
	89	ALERTMANAGER_API_SSL_VERIFY = Setting(True, [bool])
11fdf7f2 TL	90
11fdf7f2 TL	91	# iSCSI management settings
f67539c2	92	ISCSI_API_SSL_VERIFICATION = Setting(True, [bool])
11fdf7f2	93
9f95a23c TL	94	# user management settings
	95	# Time span of user passwords to expire in days.
	96	# The default value is '0' which means that user passwords are
	97	# never going to expire.
f67539c2	98	USER_PWD_EXPIRATION_SPAN = Setting(0, [int])
9f95a23c TL	99	# warning levels to notify the user that the password is going
9f95a23c TL	100	# to expire soon
f67539c2 TL	101	USER_PWD_EXPIRATION_WARNING_1 = Setting(10, [int])
f67539c2 TL	102	USER_PWD_EXPIRATION_WARNING_2 = Setting(5, [int])
9f95a23c TL	103
9f95a23c TL	104	# Password policy
f67539c2	105	PWD_POLICY_ENABLED = Setting(True, [bool])
9f95a23c	106	# Individual checks
f67539c2 TL	107	PWD_POLICY_CHECK_LENGTH_ENABLED = Setting(True, [bool])
	108	PWD_POLICY_CHECK_OLDPWD_ENABLED = Setting(True, [bool])
	109	PWD_POLICY_CHECK_USERNAME_ENABLED = Setting(False, [bool])
	110	PWD_POLICY_CHECK_EXCLUSION_LIST_ENABLED = Setting(False, [bool])
	111	PWD_POLICY_CHECK_COMPLEXITY_ENABLED = Setting(False, [bool])
	112	PWD_POLICY_CHECK_SEQUENTIAL_CHARS_ENABLED = Setting(False, [bool])
	113	PWD_POLICY_CHECK_REPETITIVE_CHARS_ENABLED = Setting(False, [bool])
9f95a23c	114	# Settings
f67539c2 TL	115	PWD_POLICY_MIN_LENGTH = Setting(8, [int])
	116	PWD_POLICY_MIN_COMPLEXITY = Setting(10, [int])
	117	PWD_POLICY_EXCLUSION_LIST = Setting(','.join(['osd', 'host', 'dashboard', 'pool',
	118	'block', 'nfs', 'ceph', 'monitors',
	119	'gateway', 'logs', 'crush', 'maps']),
	120	[str])
9f95a23c	121
aee94f69 TL	122	UNSAFE_TLS_v1_2 = Setting(False, [bool])
aee94f69 TL	123
11fdf7f2 TL	124	@staticmethod
	125	def has_default_value(name):
	126	return getattr(Settings, name, None) is None or \
f67539c2	127	getattr(Settings, name) == getattr(Options, name).default_value
11fdf7f2 TL	128
	129
	130	class SettingsMeta(type):
	131	def __getattr__(cls, attr):
f67539c2 TL	132	setting = getattr(Options, attr)
f67539c2 TL	133	return setting.cast(mgr.get_module_option(attr, setting.default_value))
11fdf7f2 TL	134
	135	def __setattr__(cls, attr, value):
	136	if not attr.startswith('_') and hasattr(Options, attr):
	137	mgr.set_module_option(attr, str(value))
	138	else:
	139	setattr(SettingsMeta, attr, value)
	140
	141	def __delattr__(cls, attr):
	142	if not attr.startswith('_') and hasattr(Options, attr):
	143	mgr.set_module_option(attr, None)
	144
	145
	146	# pylint: disable=no-init
f67539c2	147	class Settings(object, metaclass=SettingsMeta):
11fdf7f2 TL	148	pass
	149
	150
	151	def _options_command_map():
	152	def filter_attr(member):
	153	return not inspect.isroutine(member)
	154
	155	cmd_map = {}
f67539c2	156	for option, setting in inspect.getmembers(Options, filter_attr):
11fdf7f2 TL	157	if option.startswith('_'):
	158	continue
	159	key_get = 'dashboard get-{}'.format(option.lower().replace('_', '-'))
	160	key_set = 'dashboard set-{}'.format(option.lower().replace('_', '-'))
	161	key_reset = 'dashboard reset-{}'.format(option.lower().replace('_', '-'))
	162	cmd_map[key_get] = {'name': option, 'type': None}
f67539c2	163	cmd_map[key_set] = {'name': option, 'type': setting.types_as_str()}
11fdf7f2 TL	164	cmd_map[key_reset] = {'name': option, 'type': None}
	165	return cmd_map
	166
	167
	168	_OPTIONS_COMMAND_MAP = _options_command_map()
	169
	170
	171	def options_command_list():
	172	"""
	173	This function generates a list of ``get`` and ``set`` commands
	174	for each declared configuration option in class ``Options``.
	175	"""
	176	def py2ceph(pytype):
	177	if pytype == str:
	178	return 'CephString'
	179	elif pytype == int:
	180	return 'CephInt'
	181	return 'CephString'
	182
	183	cmd_list = []
	184	for cmd, opt in _OPTIONS_COMMAND_MAP.items():
	185	if cmd.startswith('dashboard get'):
	186	cmd_list.append({
	187	'cmd': '{}'.format(cmd),
	188	'desc': 'Get the {} option value'.format(opt['name']),
	189	'perm': 'r'
	190	})
	191	elif cmd.startswith('dashboard set'):
cd265ab1	192	cmd_entry = {
11fdf7f2 TL	193	'cmd': '{} name=value,type={}'
	194	.format(cmd, py2ceph(opt['type'])),
	195	'desc': 'Set the {} option value'.format(opt['name']),
	196	'perm': 'w'
cd265ab1 TL	197	}
	198	if handles_secret(cmd):
	199	cmd_entry['cmd'] = cmd
	200	cmd_entry['desc'] = '{} read from -i <file>'.format(cmd_entry['desc'])
	201	cmd_list.append(cmd_entry)
11fdf7f2 TL	202	elif cmd.startswith('dashboard reset'):
	203	desc = 'Reset the {} option to its default value'.format(
	204	opt['name'])
	205	cmd_list.append({
	206	'cmd': '{}'.format(cmd),
	207	'desc': desc,
	208	'perm': 'w'
	209	})
	210
	211	return cmd_list
	212
	213
	214	def options_schema_list():
	215	def filter_attr(member):
	216	return not inspect.isroutine(member)
	217
	218	result = []
f67539c2	219	for option, setting in inspect.getmembers(Options, filter_attr):
11fdf7f2 TL	220	if option.startswith('_'):
11fdf7f2 TL	221	continue
f67539c2 TL	222	result.append({'name': option, 'default': setting.default_value,
f67539c2 TL	223	'type': setting.types_as_str()})
11fdf7f2 TL	224
	225	return result
	226
	227
cd265ab1	228	def handle_option_command(cmd, inbuf):
11fdf7f2 TL	229	if cmd['prefix'] not in _OPTIONS_COMMAND_MAP:
	230	return -errno.ENOSYS, '', "Command not found '{}'".format(cmd['prefix'])
	231
	232	opt = _OPTIONS_COMMAND_MAP[cmd['prefix']]
	233
	234	if cmd['prefix'].startswith('dashboard reset'):
	235	delattr(Settings, opt['name'])
	236	return 0, 'Option {} reset to default value "{}"'.format(
	237	opt['name'], getattr(Settings, opt['name'])), ''
	238	elif cmd['prefix'].startswith('dashboard get'):
	239	return 0, str(getattr(Settings, opt['name'])), ''
	240	elif cmd['prefix'].startswith('dashboard set'):
cd265ab1 TL	241	if handles_secret(cmd['prefix']):
	242	value, stdout, stderr = get_secret(inbuf=inbuf)
	243	if stderr:
	244	return value, stdout, stderr
	245	else:
	246	value = cmd['value']
f67539c2 TL	247	setting = getattr(Options, opt['name'])
f67539c2 TL	248	setattr(Settings, opt['name'], setting.cast(value))
11fdf7f2	249	return 0, 'Option {} updated'.format(opt['name']), ''
cd265ab1 TL	250
	251
	252	def handles_secret(cmd: str) -> bool:
	253	return bool([cmd for secret_word in ['password', 'key'] if (secret_word in cmd)])
	254
	255
522d829b	256	@CLICheckNonemptyFileInput(desc='password/secret')
cd265ab1 TL	257	def get_secret(inbuf=None):
cd265ab1 TL	258	return inbuf, None, None