mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Add expression validation when the configuration is parsed.
This will stop Galaxy from starting with an invalid expression defined in the configuration.
This commit is contained in:
@@ -3,7 +3,10 @@ import os
|
||||
import time
|
||||
from typing import Set
|
||||
|
||||
from galaxy.exceptions import ItemAccessibilityException
|
||||
from galaxy.exceptions import (
|
||||
ConfigurationError,
|
||||
ItemAccessibilityException,
|
||||
)
|
||||
from galaxy.util.bool_expressions import (
|
||||
BooleanExpressionEvaluator,
|
||||
TokenContainedEvaluator,
|
||||
@@ -98,6 +101,7 @@ class BaseFilesSource(FilesSource):
|
||||
self.writable = kwd.pop("writable", DEFAULT_WRITABLE)
|
||||
self.requires_roles = kwd.pop("requires_roles", None)
|
||||
self.requires_groups = kwd.pop("requires_groups", None)
|
||||
self._validate_security_rules()
|
||||
# If coming from to_dict, strip API helper values
|
||||
kwd.pop("uri_root", None)
|
||||
kwd.pop("type", None)
|
||||
@@ -191,6 +195,18 @@ class BaseFilesSource(FilesSource):
|
||||
evaluator = BooleanExpressionEvaluator(token_evaluator)
|
||||
return evaluator.evaluate_expression(rule_expression)
|
||||
|
||||
def _validate_security_rules(self) -> None:
|
||||
"""Checks if the security rules defined in the plugin configuration are valid boolean expressions or raises
|
||||
a ConfigurationError exception otherwise."""
|
||||
|
||||
def _get_error_msg_for(rule_name: str) -> str:
|
||||
return f"Invalid boolean expression for '{rule_name}' in {self.label} file source plugin configuration."
|
||||
|
||||
if self.requires_roles and not BooleanExpressionEvaluator.is_valid_expression(self.requires_roles):
|
||||
raise ConfigurationError(_get_error_msg_for("requires_roles"))
|
||||
if self.requires_groups and not BooleanExpressionEvaluator.is_valid_expression(self.requires_groups):
|
||||
raise ConfigurationError(_get_error_msg_for("requires_groups"))
|
||||
|
||||
|
||||
def uri_join(*args):
|
||||
# url_join doesn't work with non-standard scheme
|
||||
|
||||
@@ -150,6 +150,17 @@ class BooleanExpressionEvaluator:
|
||||
log.error(f'BooleanExpressionEvaluator unable to evaluate expression => {expr}', exc_info=e)
|
||||
raise e
|
||||
|
||||
@classmethod
|
||||
def is_valid_expression(cls, expr: str) -> bool:
|
||||
"""Tries to evaluate the given boolean expression and returns True if it is valid or
|
||||
False if it has syntax or gramatical errors."""
|
||||
try:
|
||||
evaluator = BooleanExpressionEvaluator(ValidationOnlyTokenEvaluator())
|
||||
evaluator.evaluate_expression(expr)
|
||||
return True
|
||||
except ParseException:
|
||||
return False
|
||||
|
||||
|
||||
class TokenContainedEvaluator(TokenEvaluator):
|
||||
"""Implements the TokenEvaluator interface to determine if a token is contained
|
||||
@@ -165,3 +176,12 @@ class TokenContainedEvaluator(TokenEvaluator):
|
||||
|
||||
def evaluate(self, token: str) -> bool:
|
||||
return token in self.tokens
|
||||
|
||||
|
||||
class ValidationOnlyTokenEvaluator(TokenEvaluator):
|
||||
"""Simple TokenEvaluator that always evaluates to True for valid tokens.
|
||||
|
||||
This is only useful for validation purposes, do NOT use it for real expression evaluations."""
|
||||
|
||||
def evaluate(self, token: str) -> bool:
|
||||
return True
|
||||
|
||||
@@ -14,6 +14,27 @@ TOKEN_FORMAT = DEFAULT_TOKEN_FORMAT
|
||||
# of the *valid tokens* (those that match the TOKEN_FORMAT) will be evaluated to False.
|
||||
TOKENS_THAT_ARE_TRUE = {"T1", "token_2"}
|
||||
|
||||
VALID_EXPRESSIONS_TESTS = [
|
||||
("T1", True),
|
||||
("token_2", True),
|
||||
("T3", False),
|
||||
("valid_token", False),
|
||||
("not T3", True),
|
||||
("NOT token_2", False),
|
||||
("T1 and not T3", True),
|
||||
("NOT T1 AND token_2", False),
|
||||
("not T3 or (T3 AND token_2)", True),
|
||||
("T1 and (T3 OR token_2)", True),
|
||||
("(T3 OR T1) and not (T3 OR valid_token)", True),
|
||||
]
|
||||
|
||||
INVALID_EXPRESSIONS_TESTS = [
|
||||
"",
|
||||
"23 45",
|
||||
"'some quoted str' and not T1",
|
||||
"invalid expression",
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(scope='module')
|
||||
def contained_evaluator() -> BooleanExpressionEvaluator:
|
||||
@@ -26,30 +47,25 @@ def contained_evaluator() -> BooleanExpressionEvaluator:
|
||||
return evaluator
|
||||
|
||||
|
||||
@pytest.mark.parametrize('expr, expected', [
|
||||
("T1", True),
|
||||
("token_2", True),
|
||||
("T3", False),
|
||||
("valid_token", False),
|
||||
("not T3", True),
|
||||
("NOT token_2", False),
|
||||
("T1 and not T3", True),
|
||||
("NOT T1 AND token_2", False),
|
||||
("not T3 or (T3 AND token_2)", True),
|
||||
("T1 and (T3 OR token_2)", True),
|
||||
("(T3 OR T1) and not (T3 OR valid_token)", True),
|
||||
])
|
||||
@pytest.mark.parametrize('expr, expected', VALID_EXPRESSIONS_TESTS)
|
||||
def test_expression_evaluates_as_expected(expr: str, expected: bool, contained_evaluator: BooleanExpressionEvaluator):
|
||||
actual = contained_evaluator.evaluate_expression(expr)
|
||||
assert actual == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize('expr', [
|
||||
"",
|
||||
"23 45",
|
||||
"'some quoted str' and not T1",
|
||||
"invalid expression",
|
||||
])
|
||||
@pytest.mark.parametrize('expr', INVALID_EXPRESSIONS_TESTS)
|
||||
def test_invalid_expression_raises_exception(expr: str, contained_evaluator: BooleanExpressionEvaluator):
|
||||
with pytest.raises(Exception):
|
||||
contained_evaluator.evaluate_expression(expr)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('expr, _', VALID_EXPRESSIONS_TESTS)
|
||||
def test_is_valid_expression_return_true_when_valid(expr: str, _: bool):
|
||||
result = BooleanExpressionEvaluator.is_valid_expression(expr)
|
||||
assert result is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize('expr', INVALID_EXPRESSIONS_TESTS)
|
||||
def test_is_valid_expression_return_false_when_invalid(expr: str):
|
||||
result = BooleanExpressionEvaluator.is_valid_expression(expr)
|
||||
assert result is False
|
||||
|
||||
Reference in New Issue
Block a user