Allow expressions in job_conf resubmission conditions.

Since we allow dynamic job runners clearly job_conf.xml must be considered a trusted document. Still I know that some are uneasy with eval generally so I only allow a very safe subset of Python expressions in this job_conf document.

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)

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)

test/integration/resubmission_job_conf.xml

@@ -1,9 +1,6 @@
 <?xml version="1.0"?>
 <!-- 
     - Test resubmit without a condition.    
-    - Refactor condition check out into expression language.
-    - Add test for or to condition.
-    - Add variable any_error which is walltime_reached or memory_limit_reached or unknown_error.
     - Add variable ``attempt`` to condition language which is the resubmission count for the current destination.
     - Add variables ``time_since_queued`` and ``time_running`` to condition language.
 -->
@@ -35,6 +32,14 @@
             <resubmit condition="unknown_error" destination="retry_unknown_error" />
         </destination>
 
+        <destination id="fail_first_if_memory_or_walltime" runner="failure_runner">
+            <resubmit condition="walltime_reached or memory_limit_reached" destination="retry_test_generic" />
+        </destination>
+
+        <destination id="fail_first_any_failure" runner="failure_runner">
+            <resubmit condition="any_failure" destination="retry_test_generic" />
+        </destination>
+
         <destination id="fails_without_resubmission" runner="failure_runner">
         </destination>
 
@@ -50,6 +55,11 @@
             <param id="dest_name">retry_unknown_error</param>
         </destination>
 
+        <destination id="retry_test_generic" runner="assertion_runner">
+            <param id="dest_name">retry_test_generic</param>
+        </destination>
+
+
         <!-- Upload destination. -->
         <destination id="local" runner="local">
         </destination>

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)