AnsibleUnsafe notes

AnsibleUnsafe

ansible.utils.unsafe_proxy.AnsibleUnsafe is a mechanism to avoid template injection attacks. Values marked unsafe should not be (recursively) evaluted by Jinja2, doing so risks giving an attacker arbitrary code execution.

At time of writing (March 2024) sub-classes of AnsibleUnsafe are

• AnsibleUnsafeBytes
• AnsibleUnsafeText
• NativeJinjaUnsafeText

Values considered unsafe generally come from untrusted/external sources, e.g.

• result of any lookup {{ lookup('env', ...) }}
• result of any module executed on a target, e.g. command, template
• explicitly tagged in YAML, may_contain_braces: !unsafe "{{"

Values are usually marked by calling ansible.utils.unsafe_proxy.wrap_var(), which recursively walks sequences/mappings, replacing strings & byte strings.

Origin	Where unsafe is applied
LookupPlugin	ansible.templates.Templar._lookup()
AnsibleModule	ansible.plugins.action.ActionBase._execute_module()

Implementation details

AnsibleUnsafe objects are marked by the attribute __UNSAFE__=True. In Ansible <= 6 (ansible-core <= 2.13) casting to the base type removes it

>>> unsafe_text = AnsibleUnsafeText('abc')
>>> type(unsafe_text)
<class 'ansible.utils.unsafe_proxy.AnsibleUnsafeText'>
>>> type(str(unsafe_text))
<class 'str'>

In Ansible 7 - 9 (ansible-core 2.14 - 2.16) AnsibleUnsafeText and AnsibleUnsafeBytes override most methods, so derived values are also marked unsafe. Use AnsibleUnsafeString._strip_unsafe() instead

>>> unsafe_text = AnsibleUnsafeText('abc')
>>> type(str(unsafe_text))
<class 'ansible.utils.unsafe_proxy.AnsibleUnsafeText'>
>>> type(unsafe_text._strip_unsafe())
<class 'str'>

Investigation techniques

Is a value marked unsafe?

$ ansible localhost -e 'answer=42' -m debug -a 'msg={{ answer | type_debug }}'
localhost | SUCCESS => {
    "msg": "str"
}

$ ansible localhost -e 'env_home={{ lookup("env", "HOME") }}' \
                    -m debug -a 'msg={{ env_home | type_debug }}'
localhost | SUCCESS => {
    "msg": "AnsibleUnsafeText"
}

Where/how was a value marked unsafe?

Modify ansible/utils/unsafe_proxy.py to raise an exception given a sentinal value you can generate, e.g.

diff --git a/utils/unsafe_proxy.py b/utils/unsafe_proxy.py
index d5816ad..1e43966 100644
--- a/utils/unsafe_proxy.py
+++ b/utils/unsafe_proxy.py
@@ -201,6 +201,11 @@ class AnsibleUnsafeBytes(bytes, AnsibleUnsafe):


class AnsibleUnsafeText(str, AnsibleUnsafe):
+    def __new__(cls, *args, **kwargs):
+        s = str(*args, **kwargs)
+        if s == 'rumplestiltskin': raise RuntimeError
+        return super().__new__(cls, *args, **kwargs)
+
    def _strip_unsafe(self, /):
        return super().__str__()

Run Ansible with high verbosity (-vvv) to see the resulting traceback

$ ansible localhost -mcommand -a"echo rumplestiltskin" -vvv
...
The full traceback is:
Traceback (most recent call last):
  File ".../ansible/executor/task_executor.py", line 158, in run
    res = self._execute()
          ^^^^^^^^^^^^^^^
  File ".../ansible/executor/task_executor.py", line 633, in _execute
    result = self._handler.run(task_vars=vars_copy)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".../ansible/plugins/action/command.py", line 22, in run
    results = merge_hash(results, self._execute_module(module_name='ansible.legacy.command', task_vars=task_vars, wrap_async=wrap_async))
                                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".../ansible/plugins/action/__init__.py", line 1223, in _execute_module
    data = wrap_var(data)
          ^^^^^^^^^^^^^^
  File ".../ansible/utils/unsafe_proxy.py", line 370, in wrap_var
    v = _wrap_dict(v)
        ^^^^^^^^^^^^^
  File ".../ansible/utils/unsafe_proxy.py", line 350, in _wrap_dict
    return dict((wrap_var(k), wrap_var(item)) for k, item in v.items())
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".../ansible/utils/unsafe_proxy.py", line 350, in <genexpr>
    return dict((wrap_var(k), wrap_var(item)) for k, item in v.items())
                              ^^^^^^^^^^^^^^
  File ".../ansible/utils/unsafe_proxy.py", line 380, in wrap_var
    v = AnsibleUnsafeText(v)
        ^^^^^^^^^^^^^^^^^^^^
  File ".../ansible/utils/unsafe_proxy.py", line 213, in __new__
    if s == 'rumplestiltskin': raise RuntimeError
                              ^^^^^^^^^^^^^^^^^^
RuntimeError
...

Where/how was an unsafe marking stripped?

Use Ansible >= 7. Modify ansible/utils/unsafe_proxy.py to write a stack trace when AnsibleUnsafetext._strip_unsafe() is called with a sentinal

--- a/utils/unsafe_proxy.py
+++ b/utils/unsafe_proxy.py
@@ -55,6 +55,7 @@ __metaclass__ = type

import sys
import types
+import traceback
import warnings
from sys import intern as _sys_intern
from collections.abc import Mapping, Set
@@ -202,6 +203,9 @@ class AnsibleUnsafeBytes(bytes, AnsibleUnsafe):

class AnsibleUnsafeText(str, AnsibleUnsafe):
    def _strip_unsafe(self, /):
+        if self == 'rumplestiltskin':
+            with open('/tmp/rumplestiltskin', 'a') as f:
+                traceback.print_stack(file=f)
        return super().__str__()

    def __reduce__(self, /):

Run ansible and view the file contents

$ FOO="rumplestiltskin" v312/bin/ansible localhost -e 'foo={{ lookup("env", "FOO") }}' -m debug -a 'msg={{ foo }}'
...

$ cat /tmp/rumplestiltskin
  File "/opt/homebrew/Cellar/[email protected]/3.12.2_1/Frameworks/Python.framework/Versions/3.12/lib/python3.12/threading.py", line 1030, in _bootstrap
    self._bootstrap_inner()
  File "/opt/homebrew/Cellar/[email protected]/3.12.2_1/Frameworks/Python.framework/Versions/3.12/lib/python3.12/threading.py", line 1073, in _bootstrap_inner
    self.run()
  File "/opt/homebrew/Cellar/[email protected]/3.12.2_1/Frameworks/Python.framework/Versions/3.12/lib/python3.12/threading.py", line 1010, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/homebrew/Cellar/[email protected]/3.12.2_1/Frameworks/Python.framework/Versions/3.12/lib/python3.12/multiprocessing/queues.py", line 264, in _feed
    obj = _ForkingPickler.dumps(obj)
  File "/opt/homebrew/Cellar/[email protected]/3.12.2_1/Frameworks/Python.framework/Versions/3.12/lib/python3.12/multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
  File ".../ansible/utils/unsafe_proxy.py", line 212, in __reduce__
    return (self.__class__, (self._strip_unsafe(),))
  File ".../ansible/utils/unsafe_proxy.py", line 208, in _strip_unsafe
    traceback.print_stack(file=f)

Observations

• ansible.utils.unsafe_proxy.* is only available on the Ansible controller. The module isn't part of ansible.module_utils, so not available on targets.
• Pickling/unpickling does not strip the unsafe marker.
• JSON encoding/decoding strips the unsafe marker. This happens to AnsibleModule arguments serialised in ansible.executor.module_common._find_module_utils().

Discussion

Informed guesses of rules/design principals

• The Ansible controller is the most valued security perimeter.
• Targets are required/assumed to trust all input from the controller. By design they execute any arbitrary code that it sends.
• All templating should/does happen on the controller (80% sure).
• Ansible targets cannot/should not try to mark a value as safe or unsafe. The controller couldn't trust that determination anyway.

Ideas

Could the current dispatch table/isinstance() approach be replaced by a roundtrip through JSON? E.g.

def cast(obj):
    return json.loads(json.dumps(obj))

Would it be faster/slower?