diff --git a/lib/galaxy/jobs/runners/state_handlers/_safe_eval.py b/lib/galaxy/jobs/runners/state_handlers/_safe_eval.py new file mode 100644 index 00000000000..83480c3581c --- /dev/null +++ b/lib/galaxy/jobs/runners/state_handlers/_safe_eval.py @@ -0,0 +1,141 @@ +import re +from ast import ( + Module, + parse, + walk, +) + +AST_NODE_TYPE_WHITELIST = [ + 'Expr', 'Load', 'Str', 'Num', 'BoolOp', 'Compare', 'And', 'Eq', 'NotEq', + 'Or', 'GtE', 'LtE', 'Lt', 'Gt', 'BinOp', 'Add', 'Div', 'Sub', 'Mult', 'Mod', + 'Pow', 'LShift', 'GShift', 'BitAnd', 'BitOr', 'BitXor', 'UnaryOp', 'Invert', + 'Not', 'NotIn', 'In', 'Is', 'IsNot', 'List', 'Index', 'Subscript', + # Further checks + 'Name', 'Call', 'Attribute', +] + + +BUILTIN_AND_MATH_FUNCTIONS = 'abs|all|any|bin|chr|cmp|complex|divmod|float|hex|int|len|long|max|min|oct|ord|pow|range|reversed|round|sorted|str|sum|type|unichr|unicode|log|exp|sqrt|ceil|floor'.split('|') +STRING_AND_LIST_METHODS = [ name for name in dir('') + dir([]) if not name.startswith('_') ] +VALID_FUNCTIONS = BUILTIN_AND_MATH_FUNCTIONS + STRING_AND_LIST_METHODS + + +def _check_name(ast_node, allowed_variables=[]): + name = ast_node.id + return name in (VALID_FUNCTIONS + allowed_variables) + if re.match(r'^c\d+$', name): + return True + return name in VALID_FUNCTIONS + + +def _check_attribute( ast_node ): + attribute_name = ast_node.attr + if attribute_name not in STRING_AND_LIST_METHODS: + return False + return True + + +def _check_call( ast_node ): + # If we are calling a function or method, it better be a math, + # string or list function. + ast_func = ast_node.func + ast_func_class = ast_func.__class__.__name__ + if ast_func_class == 'Name': + if ast_func.id not in BUILTIN_AND_MATH_FUNCTIONS: + return False + elif ast_func_class == 'Attribute': + if not _check_attribute( ast_func ): + return False + else: + return False + + return True + + +def _check_expression(text, allowed_variables=[]): + """ + + >>> allowed_variables = ["c1", "c2", "c3", "c4", "c5"] + >>> _check_expression("c1", allowed_variables) + True + >>> _check_expression("eval('1+1')", allowed_variables) + False + >>> _check_expression("import sys", allowed_variables) + False + >>> _check_expression("[].__str__", allowed_variables) + False + >>> _check_expression("__builtins__", allowed_variables) + False + >>> _check_expression("'x' in globals", allowed_variables) + False + >>> _check_expression("'x' in [1,2,3]", allowed_variables) + True + >>> _check_expression("c3=='chr1' and c5>5", allowed_variables) + True + >>> _check_expression("c3=='chr1' and d5>5", allowed_variables) # Invalid d5 reference + False + >>> _check_expression("c3=='chr1' and c5>5 or exec", allowed_variables) + False + >>> _check_expression("type(c1) != type(1)", allowed_variables) + True + >>> _check_expression("c1.split(',')[1] == '1'", allowed_variables) + True + >>> _check_expression("exec 1", allowed_variables) + False + >>> _check_expression("str(c2) in [\\\"a\\\",\\\"b\\\"]", allowed_variables) + True + """ + try: + module = parse(text) + except SyntaxError: + return False + + if not isinstance(module, Module): + return False + statements = module.body + if not len(statements) == 1: + return False + expression = statements[0] + if expression.__class__.__name__ != 'Expr': + return False + + for ast_node in walk(expression): + ast_node_class = ast_node.__class__.__name__ + + # Toss out everything that is not a "simple" expression, + # imports, error handling, etc... + if ast_node_class not in AST_NODE_TYPE_WHITELIST: + return False + + # White-list more potentially dangerous types AST elements. + if ast_node_class == 'Name': + # In order to prevent loading 'exec', 'eval', etc... + # put string restriction on names allowed. + if not _check_name(ast_node, allowed_variables): + return False + # Check only valid, white-listed functions are called. + elif ast_node_class == 'Call': + if not _check_call(ast_node): + return False + # Check only valid, white-listed attributes are accessed + elif ast_node_class == 'Attribute': + if not _check_attribute(ast_node): + return False + + return True + + +def safe_eval(expression, variables): + """ + + >>> safe_eval("moo", {"moo": 5}) + 5 + >>> exception_thrown = False + >>> try: safe_eval("moo", {"cow": 5}) + ... except Exception as e: exception_thrown = True + >>> exception_thrown + True + """ + if not _check_expression(expression, allowed_variables=variables.keys()): + raise Exception("Invalid expression [%s], only a very simple subset of Python is allowed." % expression) + return eval(expression, globals(), variables) diff --git a/lib/galaxy/jobs/runners/state_handlers/resubmit.py b/lib/galaxy/jobs/runners/state_handlers/resubmit.py index 964e8d94979..9e721af8c25 100644 --- a/lib/galaxy/jobs/runners/state_handlers/resubmit.py +++ b/lib/galaxy/jobs/runners/state_handlers/resubmit.py @@ -5,6 +5,7 @@ __all__ = ('failure', ) log = logging.getLogger(__name__) from galaxy.jobs.runners import JobState +from ._safe_eval import safe_eval MESSAGES = dict( @@ -14,21 +15,40 @@ MESSAGES = dict( ) -def failure(app, job_runner, job_state): +def eval_condition(condition, job_state): runner_state = getattr(job_state, 'runner_state', None) or JobState.runner_states.UNKNOWN_ERROR if (runner_state not in (JobState.runner_states.WALLTIME_REACHED, JobState.runner_states.MEMORY_LIMIT_REACHED, JobState.runner_states.UNKNOWN_ERROR)): # not set or not a handleable runner state - return + return False + + condition_locals = { + "walltime_reached": runner_state == JobState.runner_states.WALLTIME_REACHED, + "memory_limit_reached": runner_state == JobState.runner_states.MEMORY_LIMIT_REACHED, + "unknown_error": JobState.runner_states.UNKNOWN_ERROR, + "any_failure": True, + "any_potential_job_failure": True, # Add a hook here - later on allow tools to describe things that are definitely input problems. + } + + # Small optimization to eliminate the need to parse AST and eval for simple variables. + if condition in condition_locals: + return condition_locals[condition] + else: + return safe_eval(condition, condition_locals) + + +def failure(app, job_runner, job_state): + runner_state = getattr(job_state, 'runner_state', None) or JobState.runner_states.UNKNOWN_ERROR # Intercept jobs that hit the walltime and have a walltime or # nonspecific resubmit destination configured for resubmit in job_state.job_destination.get('resubmit'): condition = resubmit.get('condition', None) - if condition and condition != runner_state: + if condition and not eval_condition(condition, job_state): # There is a resubmit defined for the destination but # its condition is not for the encountered state continue + external_id = getattr(job_state, "job_id", None) if external_id: job_log_prefix = "(%s/%s)" % (job_state.job_wrapper.job_id, job_state.job_id) diff --git a/test/integration/resubmission_job_conf.xml b/test/integration/resubmission_job_conf.xml index 2c8a735923d..3148c2cc41f 100644 --- a/test/integration/resubmission_job_conf.xml +++ b/test/integration/resubmission_job_conf.xml @@ -1,9 +1,6 @@ @@ -35,6 +32,14 @@ + + + + + + + + @@ -50,6 +55,11 @@ retry_unknown_error + + retry_test_generic + + + diff --git a/test/integration/test_job_resubmission.py b/test/integration/test_job_resubmission.py index 1587e8ce6a6..5d5e958854b 100644 --- a/test/integration/test_job_resubmission.py +++ b/test/integration/test_job_resubmission.py @@ -41,6 +41,22 @@ class JobResubmissionIntegrationTestCase(integration_util.IntegrationTestCase): def test_unknown_error(self): self._assert_job_passes(resource_parameters={"test_name": "test_unknown_error", "failure_state": "unknown_error"}) + def test_condition_expressions(self): + self._assert_job_passes(resource_parameters={"test_name": "test_condition_expressions_0", + "initial_destination": "fail_first_if_memory_or_walltime", + "failure_state": "memory_limit_reached"}) + self._assert_job_passes(resource_parameters={"test_name": "test_condition_expressions_1", + "initial_destination": "fail_first_if_memory_or_walltime", + "failure_state": "walltime_reached"}) + self._assert_job_fails(resource_parameters={"test_name": "test_condition_expressions_2", + "initial_destination": "fail_first_if_memory_or_walltime", + "failure_state": "unknown_error"}) + + def test_condition_any_failure(self): + self._assert_job_fails(resource_parameters={"test_name": "test_condition_any_failure", + "initial_destination": "fail_first_any_failure", + "failure_state": "unknown_error"}) + def _assert_job_passes(self, resource_parameters): self._run_tool_test("simple_constructs", resource_parameters=resource_parameters)