ansible_mitogen: Handle AnsibleUnsafeText et al in Ansible >= 7
Follwing fixes in Ansible 7-9 for CVE-2023-5764 cating `AnsibleUnsafeBytes` & `AnsibleUnsafeText` to `bytes()` or `str()` requires special handling. The handling is Ansible specific, so it shouldn't go in the mitogen package but rather the ansible_mitogen package. `ansible_mitogen.utils.unsafe.cast()` is most like `mitogen.utils.cast()`. During development it began as `ansible_mitogen.utils.unsafe.unwrap_var()`, closer to an inverse of `ansible.utils.unsafe_procy.wrap_var()`. Future enhancements may move in this direction. refs #977, refs #1046 See also - https://github.com/advisories/GHSA-7j69-qfc3-2fq9 - https://github.com/ansible/ansible/pull/82293 - https://github.com/mitogen-hq/mitogen/wiki/AnsibleUnsafe-notespull/1050/head
Alex Willmer
1 month ago
parent
813f253d6b
commit
b822f20007
@ -0,0 +1,79 @@
|
||||
from __future__ import absolute_import, division, print_function
|
||||
__metaclass__ = type
|
||||
|
||||
import ansible
|
||||
import ansible.utils.unsafe_proxy
|
||||
|
||||
import ansible_mitogen.utils
|
||||
|
||||
import mitogen
|
||||
import mitogen.core
|
||||
import mitogen.utils
|
||||
|
||||
__all__ = [
|
||||
'cast',
|
||||
]
|
||||
|
||||
def _cast_to_dict(obj): return {cast(k): cast(v) for k, v in obj.items()}
|
||||
def _cast_to_list(obj): return [cast(v) for v in obj]
|
||||
def _cast_unsafe(obj): return obj._strip_unsafe()
|
||||
def _passthrough(obj): return obj
|
||||
|
||||
|
||||
# A dispatch table to cast objects based on their exact type.
|
||||
# This is an optimisation, reliable fallbacks are required (e.g. isinstance())
|
||||
_CAST_DISPATCH = {
|
||||
bytes: bytes,
|
||||
dict: _cast_to_dict,
|
||||
list: _cast_to_list,
|
||||
tuple: _cast_to_list,
|
||||
mitogen.core.UnicodeType: mitogen.core.UnicodeType,
|
||||
}
|
||||
_CAST_DISPATCH.update({t: _passthrough for t in mitogen.utils.PASSTHROUGH})
|
||||
|
||||
if hasattr(ansible.utils.unsafe_proxy.AnsibleUnsafeText, '_strip_unsafe'):
|
||||
_CAST_DISPATCH.update({
|
||||
ansible.utils.unsafe_proxy.AnsibleUnsafeBytes: _cast_unsafe,
|
||||
ansible.utils.unsafe_proxy.AnsibleUnsafeText: _cast_unsafe,
|
||||
ansible.utils.unsafe_proxy.NativeJinjaUnsafeText: _cast_unsafe,
|
||||
})
|
||||
elif ansible_mitogen.utils.ansible_version[:2] <= (2, 16):
|
||||
_CAST_DISPATCH.update({
|
||||
ansible.utils.unsafe_proxy.AnsibleUnsafeBytes: bytes,
|
||||
ansible.utils.unsafe_proxy.AnsibleUnsafeText: mitogen.core.UnicodeType,
|
||||
})
|
||||
else:
|
||||
mitogen_ver = '.'.join(str(v) for v in mitogen.__version__)
|
||||
raise ImportError("Mitogen %s can't unwrap Ansible %s AnsibleUnsafe objects"
|
||||
% (mitogen_ver, ansible.__version__))
|
||||
|
||||
|
||||
def cast(obj):
|
||||
"""
|
||||
Return obj (or a copy) with subtypes of builtins cast to their supertype.
|
||||
|
||||
This is an enhanced version of :func:`mitogen.utils.cast`. In addition it
|
||||
handles ``ansible.utils.unsafe_proxy.AnsibleUnsafeText`` and variants.
|
||||
|
||||
There are types handled by :func:`ansible.utils.unsafe_proxy.wrap_var()`
|
||||
that this function currently does not handle (e.g. `set()`), or preserve
|
||||
preserve (e.g. `tuple()`). Future enhancements may change this.
|
||||
|
||||
:param obj:
|
||||
Object to undecorate.
|
||||
:returns:
|
||||
Undecorated object.
|
||||
"""
|
||||
# Fast path: obj is a known type, dispatch directly
|
||||
try:
|
||||
unwrapper = _CAST_DISPATCH[type(obj)]
|
||||
except KeyError:
|
||||
pass
|
||||
else:
|
||||
return unwrapper(obj)
|
||||
|
||||
# Slow path: obj is some unknown subclass
|
||||
if isinstance(obj, dict): return _cast_to_dict(obj)
|
||||
if isinstance(obj, (list, tuple)): return _cast_to_list(obj)
|
||||
|
||||
return mitogen.utils.cast(obj)
|
||||
@ -0,0 +1,92 @@
|
||||
import unittest
|
||||
|
||||
from ansible.utils.unsafe_proxy import AnsibleUnsafeBytes
|
||||
from ansible.utils.unsafe_proxy import AnsibleUnsafeText
|
||||
from ansible.utils.unsafe_proxy import wrap_var
|
||||
|
||||
import ansible_mitogen.utils.unsafe
|
||||
|
||||
import mitogen.core
|
||||
|
||||
|
||||
class Bytes(bytes): pass
|
||||
class Dict(dict): pass
|
||||
class List(list): pass
|
||||
class Set(set): pass
|
||||
class Text(mitogen.core.UnicodeType): pass
|
||||
class Tuple(tuple): pass
|
||||
|
||||
|
||||
class CastTest(unittest.TestCase):
|
||||
def assertIsType(self, obj, cls, msg=None):
|
||||
self.assertIs(type(obj), cls, msg)
|
||||
|
||||
def assertUnchanged(self, obj):
|
||||
self.assertIs(ansible_mitogen.utils.unsafe.cast(obj), obj)
|
||||
|
||||
def assertCasts(self, obj, expected):
|
||||
cast = ansible_mitogen.utils.unsafe.cast
|
||||
self.assertEqual(cast(obj), expected)
|
||||
self.assertIsType(cast(obj), type(expected))
|
||||
|
||||
def test_ansible_unsafe(self):
|
||||
self.assertCasts(AnsibleUnsafeBytes(b'abc'), b'abc')
|
||||
self.assertCasts(AnsibleUnsafeText(u'abc'), u'abc')
|
||||
|
||||
def test_passthrough(self):
|
||||
self.assertUnchanged(0)
|
||||
self.assertUnchanged(0.0)
|
||||
self.assertUnchanged(False)
|
||||
self.assertUnchanged(True)
|
||||
self.assertUnchanged(None)
|
||||
self.assertUnchanged(b'')
|
||||
self.assertUnchanged(u'')
|
||||
|
||||
def test_builtins_roundtrip(self):
|
||||
self.assertCasts(wrap_var(b''), b'')
|
||||
self.assertCasts(wrap_var({}), {})
|
||||
self.assertCasts(wrap_var([]), [])
|
||||
self.assertCasts(wrap_var(u''), u'')
|
||||
self.assertCasts(wrap_var(()), [])
|
||||
|
||||
def test_subtypes_roundtrip(self):
|
||||
self.assertCasts(wrap_var(Bytes()), b'')
|
||||
self.assertCasts(wrap_var(Dict()), {})
|
||||
self.assertCasts(wrap_var(List()), [])
|
||||
self.assertCasts(wrap_var(Text()), u'')
|
||||
self.assertCasts(wrap_var(Tuple()), [])
|
||||
|
||||
def test_subtype_nested_dict(self):
|
||||
obj = Dict(foo=Dict(bar=u'abc'))
|
||||
wrapped = wrap_var(obj)
|
||||
unwrapped = ansible_mitogen.utils.unsafe.cast(wrapped)
|
||||
self.assertEqual(unwrapped, {'foo': {'bar': u'abc'}})
|
||||
self.assertIsType(unwrapped, dict)
|
||||
self.assertIsType(unwrapped['foo'], dict)
|
||||
self.assertIsType(unwrapped['foo']['bar'], mitogen.core.UnicodeType)
|
||||
|
||||
def test_subtype_roundtrip_list(self):
|
||||
# wrap_var() preserves sequence types, cast() does not (for now)
|
||||
obj = List([List([u'abc'])])
|
||||
wrapped = wrap_var(obj)
|
||||
unwrapped = ansible_mitogen.utils.unsafe.cast(wrapped)
|
||||
self.assertEqual(unwrapped, [[u'abc']])
|
||||
self.assertIsType(unwrapped, list)
|
||||
self.assertIsType(unwrapped[0], list)
|
||||
self.assertIsType(unwrapped[0][0], mitogen.core.UnicodeType)
|
||||
|
||||
def test_subtype_roundtrip_tuple(self):
|
||||
# wrap_var() preserves sequence types, cast() does not (for now)
|
||||
obj = Tuple([Tuple([u'abc'])])
|
||||
wrapped = wrap_var(obj)
|
||||
unwrapped = ansible_mitogen.utils.unsafe.cast(wrapped)
|
||||
self.assertEqual(unwrapped, [[u'abc']])
|
||||
self.assertIsType(unwrapped, list)
|
||||
self.assertIsType(unwrapped[0], list)
|
||||
self.assertIsType(unwrapped[0][0], mitogen.core.UnicodeType)
|
||||
|
||||
def test_unknown_types_raise(self):
|
||||
cast = ansible_mitogen.utils.unsafe.cast
|
||||
self.assertRaises(TypeError, cast, set())
|
||||
self.assertRaises(TypeError, cast, Set())
|
||||
self.assertRaises(TypeError, cast, 4j)
|
||||
Loading…
Reference in New Issue