CVE-2014-4678: Improper Neutralization of Special Elements in Output Used by a Downstream Component ('Injection') (CWE-74)
value: The safe_eval function in Ansible before 1.6.4 does not properly restrict the code subset, which allows remote attackers to execute arbitrary code via crafted instructions. NOTE: this vulnerability exists because of an incomplete fix for CVE-2014-4657.
CWE description: The product constructs all or part of a command, data structure, or record using externally-influenced input from an upstream component, but it does not neutralize or incorrectly neutralizes special elements that could modify how it is parsed or interpreted when it is sent to a downstream component.
CVSS v3 base score: 9.8
Published: 2020-02-20T03:15Z
Repository: https://github.com/ansible/ansible
Language: Python
Files changed: lib/ansible/constants.py, lib/ansible/utils/__init__.py
Commit date: 2014-06-25 14:00:21 -0500
Commit message: Change safe_eval to a strict white list

Diff: diff --git a/lib/ansible/constants.py b/lib/ansible/constants.py
index e47d9043d..c2364f80a 100644
--- a/lib/ansible/constants.py
+++ b/lib/ansible/constants.py
@@ -29,19 +29,21 @@ def mk_boolean(value):
     if val.lower() in [ "true", "t", "y", "1", "yes" ]:
         return True
     else:
         return False
 
-def get_config(p, section, key, env_var, default, boolean=False, integer=False, floating=False):
+def get_config(p, section, key, env_var, default, boolean=False, integer=False, floating=False, islist=False):
     ''' return a configuration variable with casting '''
     value = _get_config(p, section, key, env_var, default)
     if boolean:
         return mk_boolean(value)
     if value and integer:
         return int(value)
     if value and floating:
         return float(value)
+    if value and islist:
+        return [x.strip() for x in value.split(',')]
     return value
 
 def _get_config(p, section, key, env_var, default):
     ''' helper function for get_config '''
     if env_var is not None:
@@ -127,16 +129,16 @@ DEFAULT_SUDO              = get_config(p, DEFAULTS, 'sudo', 'ANSIBLE_SUDO', Fals
 DEFAULT_SUDO_EXE          = get_config(p, DEFAULTS, 'sudo_exe', 'ANSIBLE_SUDO_EXE', 'sudo')
 DEFAULT_SUDO_FLAGS        = get_config(p, DEFAULTS, 'sudo_flags', 'ANSIBLE_SUDO_FLAGS', '-H')
 DEFAULT_HASH_BEHAVIOUR    = get_config(p, DEFAULTS, 'hash_behaviour', 'ANSIBLE_HASH_BEHAVIOUR', 'replace')
 DEFAULT_JINJA2_EXTENSIONS = get_config(p, DEFAULTS, 'jinja2_extensions', 'ANSIBLE_JINJA2_EXTENSIONS', None)
 DEFAULT_EXECUTABLE        = get_config(p, DEFAULTS, 'executable', 'ANSIBLE_EXECUTABLE', '/bin/sh')
-DEFAULT_SU_EXE = get_config(p, DEFAULTS, 'su_exe', 'ANSIBLE_SU_EXE', 'su')
-DEFAULT_SU = get_config(p, DEFAULTS, 'su', 'ANSIBLE_SU', False, boolean=True)
-DEFAULT_SU_FLAGS = get_config(p, DEFAULTS, 'su_flags', 'ANSIBLE_SU_FLAGS', '')
-DEFAULT_SU_USER = get_config(p, DEFAULTS, 'su_user', 'ANSIBLE_SU_USER', 'root')
-DEFAULT_ASK_SU_PASS = get_config(p, DEFAULTS, 'ask_su_pass', 'ANSIBLE_ASK_SU_PASS', False, boolean=True)
-DEFAULT_GATHERING = get_config(p, DEFAULTS, 'gathering', 'ANSIBLE_GATHERING', 'implicit').lower()
+DEFAULT_SU_EXE            = get_config(p, DEFAULTS, 'su_exe', 'ANSIBLE_SU_EXE', 'su')
+DEFAULT_SU                = get_config(p, DEFAULTS, 'su', 'ANSIBLE_SU', False, boolean=True)
+DEFAULT_SU_FLAGS          = get_config(p, DEFAULTS, 'su_flags', 'ANSIBLE_SU_FLAGS', '')
+DEFAULT_SU_USER           = get_config(p, DEFAULTS, 'su_user', 'ANSIBLE_SU_USER', 'root')
+DEFAULT_ASK_SU_PASS       = get_config(p, DEFAULTS, 'ask_su_pass', 'ANSIBLE_ASK_SU_PASS', False, boolean=True)
+DEFAULT_GATHERING         = get_config(p, DEFAULTS, 'gathering', 'ANSIBLE_GATHERING', 'implicit').lower()
 
 DEFAULT_ACTION_PLUGIN_PATH     = get_config(p, DEFAULTS, 'action_plugins',     'ANSIBLE_ACTION_PLUGINS', '/usr/share/ansible_plugins/action_plugins')
 DEFAULT_CALLBACK_PLUGIN_PATH   = get_config(p, DEFAULTS, 'callback_plugins',   'ANSIBLE_CALLBACK_PLUGINS', '/usr/share/ansible_plugins/callback_plugins')
 DEFAULT_CONNECTION_PLUGIN_PATH = get_config(p, DEFAULTS, 'connection_plugins', 'ANSIBLE_CONNECTION_PLUGINS', '/usr/share/ansible_plugins/connection_plugins')
 DEFAULT_LOOKUP_PLUGIN_PATH     = get_config(p, DEFAULTS, 'lookup_plugins',     'ANSIBLE_LOOKUP_PLUGINS', '/usr/share/ansible_plugins/lookup_plugins')
@@ -150,10 +152,11 @@ ANSIBLE_NOCOWS                 = get_config(p, DEFAULTS, 'nocows', 'ANSIBLE_NOCO
 DISPLAY_SKIPPED_HOSTS          = get_config(p, DEFAULTS, 'display_skipped_hosts', 'DISPLAY_SKIPPED_HOSTS', True, boolean=True)
 DEFAULT_UNDEFINED_VAR_BEHAVIOR = get_config(p, DEFAULTS, 'error_on_undefined_vars', 'ANSIBLE_ERROR_ON_UNDEFINED_VARS', True, boolean=True)
 HOST_KEY_CHECKING              = get_config(p, DEFAULTS, 'host_key_checking',  'ANSIBLE_HOST_KEY_CHECKING',    True, boolean=True)
 SYSTEM_WARNINGS                = get_config(p, DEFAULTS, 'system_warnings', 'ANSIBLE_SYSTEM_WARNINGS', True, boolean=True)
 DEPRECATION_WARNINGS           = get_config(p, DEFAULTS, 'deprecation_warnings', 'ANSIBLE_DEPRECATION_WARNINGS', True, boolean=True)
+DEFAULT_CALLABLE_WHITELIST     = get_config(p, DEFAULTS, 'callable_whitelist', 'ANSIBLE_CALLABLE_WHITELIST', [], islist=True)
 
 # CONNECTION RELATED
 ANSIBLE_SSH_ARGS               = get_config(p, 'ssh_connection', 'ssh_args', 'ANSIBLE_SSH_ARGS', None)
 ANSIBLE_SSH_CONTROL_PATH       = get_config(p, 'ssh_connection', 'control_path', 'ANSIBLE_SSH_CONTROL_PATH', "%(directory)s/ansible-ssh-%%h-%%p-%%r")
 ANSIBLE_SSH_PIPELINING         = get_config(p, 'ssh_connection', 'pipelining', 'ANSIBLE_SSH_PIPELINING', False, boolean=True)
diff --git a/lib/ansible/utils/__init__.py b/lib/ansible/utils/__init__.py
index e3ad20ad8..89b8fc75d 100644
--- a/lib/ansible/utils/__init__.py
+++ b/lib/ansible/utils/__init__.py
@@ -1063,25 +1063,35 @@ def safe_eval(expr, locals={}, include_exceptions=False):
             set(
                 (ast.Set,)
             )
         )
 
-    # builtin functions that are not safe to call
-    INVALID_CALLS = (
-       'classmethod', 'compile', 'delattr', 'eval', 'execfile', 'file',
-       'filter', 'help', 'input', 'object', 'open', 'raw_input', 'reduce',
-       'reload', 'repr', 'setattr', 'staticmethod', 'super', 'type',
-    )
+    # builtin functions that are safe to call
+    BUILTIN_WHITELIST = [
+        'abs', 'all', 'any', 'basestring', 'bin', 'bool', 'buffer', 'bytearray',
+        'bytes', 'callable', 'chr', 'cmp', 'coerce', 'complex', 'copyright', 'credits',
+        'dict', 'dir', 'divmod', 'enumerate', 'exit', 'float', 'format', 'frozenset',
+        'getattr', 'globals', 'hasattr', 'hash', 'hex', 'id', 'int', 'intern',
+        'isinstance', 'issubclass', 'iter', 'len', 'license', 'list', 'locals', 'long',
+        'map', 'max', 'memoryview', 'min', 'next', 'oct', 'ord', 'pow', 'print',
+        'property', 'quit', 'range', 'reversed', 'round', 'set', 'slice', 'sorted',
+        'str', 'sum', 'tuple', 'unichr', 'unicode', 'vars', 'xrange', 'zip',
+    ]
+
+    filter_list = []
+    for filter in filter_loader.all():
+        filter_list.extend(filter.filters().keys())
+
+    CALL_WHITELIST = BUILTIN_WHITELIST + filter_list + C.DEFAULT_CALLABLE_WHITELIST
 
     class CleansingNodeVisitor(ast.NodeVisitor):
         def generic_visit(self, node):
             if type(node) not in SAFE_NODES:
-                #raise Exception("invalid expression (%s) type=%s" % (expr, type(node)))
                 raise Exception("invalid expression (%s)" % expr)
             super(CleansingNodeVisitor, self).generic_visit(node)
         def visit_Call(self, call):
-            if call.func.id in INVALID_CALLS:
+            if call.func.id not in CALL_WHITELIST:
                 raise Exception("invalid function: %s" % call.func.id)
 
     if not isinstance(expr, basestring):
         # already templated to a datastructure, perhaps?
         if include_exceptions:

Vulnerable code: def get_config(p, section, key, env_var, default, boolean=False, integer=False, floating=False):
DEFAULT_SU_EXE = get_config(p, DEFAULTS, 'su_exe', 'ANSIBLE_SU_EXE', 'su')
DEFAULT_SU = get_config(p, DEFAULTS, 'su', 'ANSIBLE_SU', False, boolean=True)
DEFAULT_SU_FLAGS = get_config(p, DEFAULTS, 'su_flags', 'ANSIBLE_SU_FLAGS', '')
DEFAULT_SU_USER = get_config(p, DEFAULTS, 'su_user', 'ANSIBLE_SU_USER', 'root')
DEFAULT_ASK_SU_PASS = get_config(p, DEFAULTS, 'ask_su_pass', 'ANSIBLE_ASK_SU_PASS', False, boolean=True)
DEFAULT_GATHERING = get_config(p, DEFAULTS, 'gathering', 'ANSIBLE_GATHERING', 'implicit').lower()
    # builtin functions that are not safe to call
    INVALID_CALLS = (
       'classmethod', 'compile', 'delattr', 'eval', 'execfile', 'file',
       'filter', 'help', 'input', 'object', 'open', 'raw_input', 'reduce',
       'reload', 'repr', 'setattr', 'staticmethod', 'super', 'type',
    )
                #raise Exception("invalid expression (%s) type=%s" % (expr, type(node)))
            if call.func.id in INVALID_CALLS:
Fixed code: def get_config(p, section, key, env_var, default, boolean=False, integer=False, floating=False, islist=False):
    if value and islist:
        return [x.strip() for x in value.split(',')]
DEFAULT_SU_EXE            = get_config(p, DEFAULTS, 'su_exe', 'ANSIBLE_SU_EXE', 'su')
DEFAULT_SU                = get_config(p, DEFAULTS, 'su', 'ANSIBLE_SU', False, boolean=True)
DEFAULT_SU_FLAGS          = get_config(p, DEFAULTS, 'su_flags', 'ANSIBLE_SU_FLAGS', '')
DEFAULT_SU_USER           = get_config(p, DEFAULTS, 'su_user', 'ANSIBLE_SU_USER', 'root')
DEFAULT_ASK_SU_PASS       = get_config(p, DEFAULTS, 'ask_su_pass', 'ANSIBLE_ASK_SU_PASS', False, boolean=True)
DEFAULT_GATHERING         = get_config(p, DEFAULTS, 'gathering', 'ANSIBLE_GATHERING', 'implicit').lower()
DEFAULT_CALLABLE_WHITELIST     = get_config(p, DEFAULTS, 'callable_whitelist', 'ANSIBLE_CALLABLE_WHITELIST', [], islist=True)
    # builtin functions that are safe to call
    BUILTIN_WHITELIST = [
        'abs', 'all', 'any', 'basestring', 'bin', 'bool', 'buffer', 'bytearray',
        'bytes', 'callable', 'chr', 'cmp', 'coerce', 'complex', 'copyright', 'credits',
        'dict', 'dir', 'divmod', 'enumerate', 'exit', 'float', 'format', 'frozenset',
        'getattr', 'globals', 'hasattr', 'hash', 'hex', 'id', 'int', 'intern',
        'isinstance', 'issubclass', 'iter', 'len', 'license', 'list', 'locals', 'long',
        'map', 'max', 'memoryview', 'min', 'next', 'oct', 'ord', 'pow', 'print',
        'property', 'quit', 'range', 'reversed', 'round', 'set', 'slice', 'sorted',
        'str', 'sum', 'tuple', 'unichr', 'unicode', 'vars', 'xrange', 'zip',
    ]

    filter_list = []
    for filter in filter_loader.all():
        filter_list.extend(filter.filters().keys())

    CALL_WHITELIST = BUILTIN_WHITELIST + filter_list + C.DEFAULT_CALLABLE_WHITELIST
            if call.func.id not in CALL_WHITELIST: