Wrap CLI Passwords with AnsibleUnsafeText, ensure unsafe context is not lost during encode/decode (#63351)

* Wrap .encode and .decode on AnsibleUnsafe objects

* runme.sh needs to be executable

* ci_complete

* Update changelog with CVE

changelogs/fragments/dont-template-cli-passwords.yml

@@ -0,0 +1,12 @@
+bugfixes:
+- >
+  **security issue** - Convert CLI provided passwords to text initially, to
+  prevent unsafe context being lost when converting from bytes->text during
+  post processing of PlayContext. This prevents CLI provided passwords from
+  being incorrectly templated (CVE-2019-14856)
+- >
+  **security issue** - Update ``AnsibleUnsafeText`` and ``AnsibleUnsafeBytes``
+  to maintain unsafe context by overriding ``.encode`` and ``.decode``. This
+  prevents future issues with ``to_text``, ``to_bytes``, or ``to_native``
+  removing the unsafe wrapper when converting between string types
+  (CVE-2019-14856)

lib/ansible/cli/__init__.py

@@ -29,7 +29,7 @@ from ansible.release import __version__
 from ansible.utils.collection_loader import AnsibleCollectionLoader, get_collection_name_from_path, set_collection_playbook_paths
 from ansible.utils.display import Display
 from ansible.utils.path import unfrackpath
-from ansible.utils.unsafe_proxy import AnsibleUnsafeBytes
+from ansible.utils.unsafe_proxy import to_unsafe_text
 from ansible.vars.manager import VariableManager
 
 try:
@@ -240,8 +240,6 @@ class CLI(with_metaclass(ABCMeta, object)):
             if op['ask_pass']:
                 sshpass = getpass.getpass(prompt="SSH password: ")
                 become_prompt = "%s password[defaults to SSH password]: " % become_prompt_method
-                if sshpass:
-                    sshpass = to_bytes(sshpass, errors='strict', nonstring='simplerepr')
             else:
                 become_prompt = "%s password: " % become_prompt_method
 
@@ -249,17 +247,15 @@ class CLI(with_metaclass(ABCMeta, object)):
                 becomepass = getpass.getpass(prompt=become_prompt)
                 if op['ask_pass'] and becomepass == '':
                     becomepass = sshpass
-                if becomepass:
-                    becomepass = to_bytes(becomepass)
         except EOFError:
             pass
 
         # we 'wrap' the passwords to prevent templating as
         # they can contain special chars and trigger it incorrectly
         if sshpass:
-            sshpass = AnsibleUnsafeBytes(sshpass)
+            sshpass = to_unsafe_text(sshpass)
         if becomepass:
-            becomepass = AnsibleUnsafeBytes(becomepass)
+            becomepass = to_unsafe_text(becomepass)
 
         return (sshpass, becomepass)
 

lib/ansible/template/__init__.py

@@ -272,7 +272,7 @@ class AnsibleContext(Context):
             for item in val:
                 if self._is_unsafe(item):
                     return True
-        elif isinstance(val, string_types) and hasattr(val, '__UNSAFE__'):
+        elif hasattr(val, '__UNSAFE__'):
             return True
         return False
 

lib/ansible/utils/unsafe_proxy.py

@@ -66,11 +66,15 @@ class AnsibleUnsafe(object):
 
 
 class AnsibleUnsafeBytes(binary_type, AnsibleUnsafe):
-    pass
+    def decode(self, *args, **kwargs):
+        """Wrapper method to ensure type conversions maintain unsafe context"""
+        return AnsibleUnsafeText(super(AnsibleUnsafeBytes, self).decode(*args, **kwargs))
 
 
 class AnsibleUnsafeText(text_type, AnsibleUnsafe):
-    pass
+    def encode(self, *args, **kwargs):
+        """Wrapper method to ensure type conversions maintain unsafe context"""
+        return AnsibleUnsafeBytes(super(AnsibleUnsafeText, self).encode(*args, **kwargs))
 
 
 class UnsafeProxy(object):

test/integration/targets/cli/aliases

@@ -0,0 +1,2 @@
+needs/target/setup_pexpect
+shippable/posix/group3

test/integration/targets/cli/runme.sh

@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+
+set -eux
+
+ANSIBLE_ROLES_PATH=../ ansible-playbook setup.yml
+
+python test-cli.py

test/integration/targets/cli/setup.yml

@@ -0,0 +1,4 @@
+- hosts: localhost
+  gather_facts: no
+  roles:
+    - setup_pexpect

test/integration/targets/cli/test-cli.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python
+# Copyright (c) 2019 Matt Martz <matt@sivel.net>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import os
+
+import pexpect
+
+os.environ['ANSIBLE_NOCOLOR'] = '1'
+out = pexpect.run(
+    'ansible localhost -m debug -a msg="{{ ansible_password }}" -k',
+    events={
+        'SSH password:': '{{ 1 + 2 }}\n'
+    }
+)
+
+assert b'{{ 1 + 2 }}' in out

test/units/module_utils/test_text.py

@@ -16,6 +16,7 @@ from ansible.module_utils.six import PY3
 # Internal API while this is still being developed.  Eventually move to
 # module_utils.common.text
 from ansible.module_utils._text import to_text, to_bytes, to_native
+from ansible.utils.unsafe_proxy import AnsibleUnsafeBytes, AnsibleUnsafeText
 
 
 # Format: byte representation, text representation, encoding of byte representation
@@ -51,3 +52,13 @@ def test_to_bytes(in_string, encoding, expected):
 def test_to_native(in_string, encoding, expected):
     """test happy path of encoding to native strings"""
     assert to_native(in_string, encoding) == expected
+
+
+def test_to_text_unsafe():
+    assert isinstance(to_text(AnsibleUnsafeBytes(b'foo')), AnsibleUnsafeText)
+    assert to_text(AnsibleUnsafeBytes(b'foo')) == AnsibleUnsafeText(u'foo')
+
+
+def test_to_bytes_unsafe():
+    assert isinstance(to_bytes(AnsibleUnsafeText(u'foo')), AnsibleUnsafeBytes)
+    assert to_bytes(AnsibleUnsafeText(u'foo')) == AnsibleUnsafeBytes(b'foo')

test/units/playbook/test_base.py

@@ -26,6 +26,7 @@ from ansible.module_utils.six import string_types
 from ansible.playbook.attribute import FieldAttribute
 from ansible.template import Templar
 from ansible.playbook import base
+from ansible.utils.unsafe_proxy import AnsibleUnsafeBytes, AnsibleUnsafeText
 
 from units.mock.loader import DictDataLoader
 
@@ -620,3 +621,12 @@ class TestBaseSubClass(TestBase):
         ds = {'test_attr_method_missing': a_string}
         bsc = self._base_validate(ds)
         self.assertEquals(bsc.test_attr_method_missing, a_string)
+
+    def test_get_validated_value_string_rewrap_unsafe(self):
+        attribute = FieldAttribute(isa='string')
+        value = AnsibleUnsafeText(u'bar')
+        templar = Templar(None)
+        bsc = self.ClassUnderTest()
+        result = bsc.get_validated_value('foo', attribute, value, templar)
+        self.assertIsInstance(result, AnsibleUnsafeText)
+        self.assertEquals(result, AnsibleUnsafeText(u'bar'))