github/plexpy
1
0
Fork 0
mirror of https://github.com/Tautulli/Tautulli.git synced 2025-08-19 21:03:21 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Validate notifier custom conditions before saving

Browse source 
* Fixes #1846
This commit is contained in:
JonnyWong16 2022-10-10 19:05:13 +00:00 • committed by GitHub
commit e6829310ad
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 84 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

74
plexpy/notifiers.py
View file

@ -112,7 +112,12 @@ AGENT_IDS = {'growl': 0,
'gotify': 29
}
DEFAULT_CUSTOM_CONDITIONS = [{'parameter': '', 'operator': '', 'value': ''}]
DEFAULT_CUSTOM_CONDITIONS = [{'parameter': '', 'operator': '', 'value': [], 'type': None}]
CUSTOM_CONDITION_TYPE_OPERATORS = {
'float': ['is', 'is not', 'is greater than', 'is less than'],
'int': ['is', 'is not', 'is greater than', 'is less than'],
'str': ['contains', 'does not contain', 'is', 'is not', 'begins with', 'does not begin with', 'ends with', 'does not end with'],
}
def available_notification_agents():
@ -642,13 +647,18 @@ def set_notifier_config(notifier_id=None, agent_id=None, **kwargs):
agent_class = get_agent_class(agent_id=agent['id'], config=notifier_config)
custom_conditions = validate_conditions(kwargs.get('custom_conditions'))
if custom_conditions is False:
logger.error("Tautulli Notifiers :: Unable to update notification agent: Invalid custom conditions.")
return False
keys = {'id': notifier_id}
values = {'agent_id': agent['id'],
'agent_name': agent['name'],
'agent_label': agent['label'],
'friendly_name': kwargs.get('friendly_name', ''),
'notifier_config': json.dumps(agent_class.config),
'custom_conditions': kwargs.get('custom_conditions', json.dumps(DEFAULT_CUSTOM_CONDITIONS)),
'custom_conditions': json.dumps(custom_conditions or DEFAULT_CUSTOM_CONDITIONS),
'custom_conditions_logic': kwargs.get('custom_conditions_logic', ''),
}
values.update(actions)
@ -685,6 +695,66 @@ def send_notification(notifier_id=None, subject='', body='', notify_action='', n
logger.debug("Tautulli Notifiers :: Notification requested but no notifier_id received.")
def validate_conditions(custom_conditions):
if custom_conditions is None:
return DEFAULT_CUSTOM_CONDITIONS
try:
conditions = json.loads(custom_conditions)
except ValueError:
logger.error("Tautulli Notifiers :: Unable to parse custom conditions json: %s" % custom_conditions)
return False
if not isinstance(conditions, list):
logger.error("Tautulli Notifiers :: Invalid custom conditions: %s. Conditions must be a list." % conditions)
return False
validated_conditions = []
for condition in conditions:
validated_condition = DEFAULT_CUSTOM_CONDITIONS[0].copy()
if not isinstance(condition, dict):
logger.error("Tautulli Notifiers :: Invalid custom condition: %s. Condition must be a dict." % condition)
return False
parameter = str(condition.get('parameter', '')).lower()
operator = str(condition.get('operator', '')).lower()
values = condition.get('value', [])
if parameter:
parameter_type = common.NOTIFICATION_PARAMETERS_TYPES.get(parameter)
if not parameter_type:
logger.error("Tautulli Notifiers :: Invalid parameter '%s' in custom condition: %s" % (parameter, condition))
return False
validated_condition['parameter'] = parameter.lower()
validated_condition['type'] = parameter_type
if operator:
if operator not in CUSTOM_CONDITION_TYPE_OPERATORS.get(parameter_type, []):
logger.error("Tautulli Notifiers :: Invalid operator '%s' for parameter '%s' in custom condition: %s" % (operator, parameter, condition))
return False
validated_condition['operator'] = operator
if values:
if not isinstance(values, list):
values = [values]
for value in values:
if not isinstance(value, (str, int, float)):
logger.error("Tautulli Notifiers :: Invalid value '%s' for parameter '%s' in custom condition: %s" % (value, parameter, condition))
return False
validated_condition['value'] = values
validated_conditions.append(validated_condition)
return validated_conditions
def blacklist_logger():
db = database.MonitorDatabase()
notifiers = db.select('SELECT notifier_config FROM notifiers')