Ensure that unsafe is more difficult to lose [stable-2.16] (#82293)

* Ensure that unsafe is more difficult to lose

* Add Task.untemplated_args, and switch assert over to use it
* Don't use re in first_found, switch to using native string methods
* If nested templating results in unsafe, just error, don't continue

* ci_complete

changelogs/fragments/cve-2023-5764.yml

@@ -0,0 +1,6 @@
+security_fixes:
+- templating - Address issues where internal templating can cause unsafe
+  variables to lose their unsafe designation (CVE-2023-5764)
+breaking_changes:
+- assert - Nested templating may result in an inability for the conditional
+  to be evaluated. See the porting guide for more information.

lib/ansible/module_utils/common/json.py

@@ -30,7 +30,7 @@ def _preprocess_unsafe_encode(value):
     Used in ``AnsibleJSONEncoder.iterencode``
     """
     if _is_unsafe(value):
-        value = {'__ansible_unsafe': to_text(value, errors='surrogate_or_strict', nonstring='strict')}
+        value = {'__ansible_unsafe': to_text(value._strip_unsafe(), errors='surrogate_or_strict', nonstring='strict')}
     elif is_sequence(value):
         value = [_preprocess_unsafe_encode(v) for v in value]
     elif isinstance(value, Mapping):
@@ -63,7 +63,7 @@ def default(self, o):
                 value = {'__ansible_vault': to_text(o._ciphertext, errors='surrogate_or_strict', nonstring='strict')}
         elif getattr(o, '__UNSAFE__', False):
             # unsafe object, this will never be triggered, see ``AnsibleJSONEncoder.iterencode``
-            value = {'__ansible_unsafe': to_text(o, errors='surrogate_or_strict', nonstring='strict')}
+            value = {'__ansible_unsafe': to_text(o._strip_unsafe(), errors='surrogate_or_strict', nonstring='strict')}
         elif isinstance(o, Mapping):
             # hostvars and other objects
             value = dict(o)

lib/ansible/parsing/yaml/dumper.py

@@ -24,7 +24,7 @@
 from ansible.module_utils.six import text_type, binary_type
 from ansible.module_utils.common.yaml import SafeDumper
 from ansible.parsing.yaml.objects import AnsibleUnicode, AnsibleSequence, AnsibleMapping, AnsibleVaultEncryptedUnicode
-from ansible.utils.unsafe_proxy import AnsibleUnsafeText, AnsibleUnsafeBytes, NativeJinjaUnsafeText, NativeJinjaText
+from ansible.utils.unsafe_proxy import AnsibleUnsafeText, AnsibleUnsafeBytes, NativeJinjaUnsafeText, NativeJinjaText, _is_unsafe
 from ansible.template import AnsibleUndefined
 from ansible.vars.hostvars import HostVars, HostVarsVars
 from ansible.vars.manager import VarsWithSources
@@ -47,10 +47,14 @@ def represent_vault_encrypted_unicode(self, data):
 
 
 def represent_unicode(self, data):
+    if _is_unsafe(data):
+        data = data._strip_unsafe()
     return yaml.representer.SafeRepresenter.represent_str(self, text_type(data))
 
 
 def represent_binary(self, data):
+    if _is_unsafe(data):
+        data = data._strip_unsafe()
     return yaml.representer.SafeRepresenter.represent_binary(self, binary_type(data))
 
 

lib/ansible/playbook/conditional.py

@@ -21,7 +21,7 @@
 
 import typing as t
 
-from ansible.errors import AnsibleError, AnsibleUndefinedVariable
+from ansible.errors import AnsibleError, AnsibleUndefinedVariable, AnsibleTemplateError
 from ansible.module_utils.common.text.converters import to_native
 from ansible.playbook.attribute import FieldAttribute
 from ansible.template import Templar
@@ -102,14 +102,14 @@ def _check_conditional(self, conditional: str, templar: Templar, all_vars: dict[
                     return False
 
             # If the result of the first-pass template render (to resolve inline templates) is marked unsafe,
-            # explicitly disable lookups on the final pass to prevent evaluation of untrusted content in the
-            # constructed template.
-            disable_lookups = hasattr(conditional, '__UNSAFE__')
+            # explicitly fail since the next templating operation would never evaluate
+            if hasattr(conditional, '__UNSAFE__'):
+                raise AnsibleTemplateError('Conditional is marked as unsafe, and cannot be evaluated.')
 
             # NOTE The spaces around True and False are intentional to short-circuit literal_eval for
             #      jinja2_native=False and avoid its expensive calls.
             return templar.template(
                 "{%% if %s %%} True {%% else %%} False {%% endif %%}" % conditional,
-                disable_lookups=disable_lookups).strip() == "True"
+            ).strip() == "True"
         except AnsibleUndefinedVariable as e:
             raise AnsibleUndefinedVariable("error while evaluating conditional (%s): %s" % (original, e))

lib/ansible/playbook/task.py

@@ -289,6 +289,30 @@ def post_validate(self, templar):
 
         super(Task, self).post_validate(templar)
 
+    def _post_validate_args(self, attr, value, templar):
+        # smuggle an untemplated copy of the task args for actions that need more control over the templating of their
+        # input (eg, debug's var/msg, assert's "that" conditional expressions)
+        self.untemplated_args = value
+
+        # now recursively template the args dict
+        args = templar.template(value)
+
+        # FIXME: could we just nuke this entirely and/or wrap it up in ModuleArgsParser or something?
+        if '_variable_params' in args:
+            variable_params = args.pop('_variable_params')
+            if isinstance(variable_params, dict):
+                if C.INJECT_FACTS_AS_VARS:
+                    display.warning("Using a variable for a task's 'args' is unsafe in some situations "
+                                    "(see https://docs.ansible.com/ansible/devel/reference_appendices/faq.html#argsplat-unsafe)")
+                variable_params.update(args)
+                args = variable_params
+            else:
+                # if we didn't get a dict, it means there's garbage remaining after k=v parsing, just give up
+                # see https://github.com/ansible/ansible/issues/79862
+                raise AnsibleError(f"invalid or malformed argument: '{variable_params}'")
+
+        return args
+
     def _post_validate_loop(self, attr, value, templar):
         '''
         Override post validation for the loop field, which is templated

lib/ansible/plugins/action/assert.py

@@ -64,8 +64,29 @@ def run(self, tmp=None, task_vars=None):
 
         quiet = boolean(self._task.args.get('quiet', False), strict=False)
 
+        # directly access 'that' via untemplated args from the task so we can intelligently trust embedded
+        # templates and preserve the original inputs/locations for better messaging on assert failures and
+        # errors.
+        # FIXME: even in devel, things like `that: item` don't always work properly (truthy string value
+        # is not really an embedded expression)
+        # we could fix that by doing direct var lookups on the inputs
+        # FIXME: some form of this code should probably be shared between debug, assert, and
+        # Task.post_validate, since they
+        # have a lot of overlapping needs
+        try:
+            thats = self._task.untemplated_args['that']
+        except KeyError:
+            # in the case of "we got our entire args dict from a template", we can just consult the
+            # post-templated dict (the damage has likely already been done for embedded templates anyway)
+            thats = self._task.args['that']
+
+        # FIXME: this is a case where we only want to resolve indirections, NOT recurse containers
+        # (and even then, the leaf-most expression being wrapped is at least suboptimal
+        # (since its expression will be "eaten").
+        if isinstance(thats, str):
+            thats = self._templar.template(thats)
+
         # make sure the 'that' items are a list
-        thats = self._task.args['that']
         if not isinstance(thats, list):
             thats = [thats]
 

lib/ansible/plugins/callback/__init__.py

@@ -38,7 +38,7 @@
 from ansible.plugins import AnsiblePlugin
 from ansible.utils.color import stringc
 from ansible.utils.display import Display
-from ansible.utils.unsafe_proxy import AnsibleUnsafeText, NativeJinjaUnsafeText
+from ansible.utils.unsafe_proxy import AnsibleUnsafeText, NativeJinjaUnsafeText, _is_unsafe
 from ansible.vars.clean import strip_internal_keys, module_response_deepcopy
 
 import yaml
@@ -113,6 +113,8 @@ def _munge_data_for_lossy_yaml(scalar):
 
 def _pretty_represent_str(self, data):
     """Uses block style for multi-line strings"""
+    if _is_unsafe(data):
+        data = data._strip_unsafe()
     data = text_type(data)
     if _should_use_block(data):
         style = '|'

lib/ansible/plugins/filter/core.py

@@ -37,6 +37,7 @@
 from ansible.utils.encrypt import do_encrypt, PASSLIB_AVAILABLE
 from ansible.utils.hashing import md5s, checksum_s
 from ansible.utils.unicode import unicode_wrap
+from ansible.utils.unsafe_proxy import _is_unsafe
 from ansible.utils.vars import merge_hash
 
 display = Display()
@@ -215,6 +216,8 @@ def from_yaml(data):
         # The ``text_type`` call here strips any custom
         # string wrapper class, so that CSafeLoader can
         # read the data
+        if _is_unsafe(data):
+            data = data._strip_unsafe()
         return yaml_load(text_type(to_text(data, errors='surrogate_or_strict')))
     return data
 
@@ -224,6 +227,8 @@ def from_yaml_all(data):
         # The ``text_type`` call here strips any custom
         # string wrapper class, so that CSafeLoader can
         # read the data
+        if _is_unsafe(data):
+            data = data._strip_unsafe()
         return yaml_load_all(text_type(to_text(data, errors='surrogate_or_strict')))
     return data
 

lib/ansible/plugins/lookup/first_found.py

@@ -139,7 +139,6 @@
     elements: path
 """
 import os
-import re
 
 from collections.abc import Mapping, Sequence
 
@@ -150,10 +149,22 @@
 from ansible.plugins.lookup import LookupBase
 
 
+def _splitter(value, chars):
+    chars = set(chars)
+    v = ''
+    for c in value:
+        if c in chars:
+            yield v
+            v = ''
+            continue
+        v += c
+    yield v
+
+
 def _split_on(terms, spliters=','):
     termlist = []
     if isinstance(terms, string_types):
-        termlist = re.split(r'[%s]' % ''.join(map(re.escape, spliters)), terms)
+        termlist = list(_splitter(terms, spliters))
     else:
         # added since options will already listify
         for t in terms:

lib/ansible/template/__init__.py

@@ -31,7 +31,7 @@
 from numbers import Number
 from traceback import format_exc
 
-from jinja2.exceptions import TemplateSyntaxError, UndefinedError
+from jinja2.exceptions import TemplateSyntaxError, UndefinedError, SecurityError
 from jinja2.loaders import FileSystemLoader
 from jinja2.nativetypes import NativeEnvironment
 from jinja2.runtime import Context, StrictUndefined
@@ -55,7 +55,7 @@
 from ansible.utils.display import Display
 from ansible.utils.listify import listify_lookup_plugin_terms
 from ansible.utils.native_jinja import NativeJinjaText
-from ansible.utils.unsafe_proxy import to_unsafe_text, wrap_var
+from ansible.utils.unsafe_proxy import to_unsafe_text, wrap_var, AnsibleUnsafeText, AnsibleUnsafeBytes, NativeJinjaUnsafeText
 
 display = Display()
 
@@ -365,10 +365,21 @@ class AnsibleContext(Context):
     flag is checked post-templating, and (when set) will result in the
     final templated result being wrapped in AnsibleUnsafe.
     '''
+    _disallowed_callables = frozenset({
+        AnsibleUnsafeText._strip_unsafe.__qualname__,
+        AnsibleUnsafeBytes._strip_unsafe.__qualname__,
+        NativeJinjaUnsafeText._strip_unsafe.__qualname__,
+    })
+
     def __init__(self, *args, **kwargs):
         super(AnsibleContext, self).__init__(*args, **kwargs)
         self.unsafe = False
 
+    def call(self, obj, *args, **kwargs):
+        if getattr(obj, '__qualname__', None) in self._disallowed_callables or obj in self._disallowed_callables:
+            raise SecurityError(f"{obj!r} is not safely callable")
+        return super().call(obj, *args, **kwargs)
+
     def _is_unsafe(self, val):
         '''
         Our helper function, which will also recursively check dict and