Merge pull request #12294 from ansible/fix-password-lookup

Fix problem with "=" in the initial file path.

lib/ansible/plugins/lookup/password.py

@@ -36,6 +36,54 @@ DEFAULT_LENGTH = 20
 VALID_PARAMS = frozenset(('length', 'encrypt', 'chars'))
 
 
+def _parse_parameters(term):
+    # Hacky parsing of params
+    # See https://github.com/ansible/ansible-modules-core/issues/1968#issuecomment-136842156
+    # and the first_found lookup For how we want to fix this later
+    first_split = term.split(' ', 1)
+    if len(first_split) <= 1:
+        # Only a single argument given, therefore it's a path
+        relpath = term
+        params = dict()
+    else:
+        relpath = first_split[0]
+        params = parse_kv(first_split[1])
+        if '_raw_params' in params:
+            # Spaces in the path?
+            relpath = ' '.join((relpath, params['_raw_params']))
+            del params['_raw_params']
+
+            # Check that we parsed the params correctly
+            if not term.startswith(relpath):
+                # Likely, the user had a non parameter following a parameter.
+                # Reject this as a user typo
+                raise AnsibleError('Unrecognized value after key=value parameters given to password lookup')
+        # No _raw_params means we already found the complete path when
+        # we split it initially
+
+    # Check for invalid parameters.  Probably a user typo
+    invalid_params = frozenset(params.keys()).difference(VALID_PARAMS)
+    if invalid_params:
+        raise AnsibleError('Unrecognized parameter(s) given to password lookup: %s' % ', '.join(invalid_params))
+
+    # Set defaults
+    params['length'] = int(params.get('length', DEFAULT_LENGTH))
+    params['encrypt'] = params.get('encrypt', None)
+
+    params['chars'] = params.get('chars', None)
+    if params['chars']:
+        tmp_chars = []
+        if ',,' in params['chars']:
+            tmp_chars.append(u',')
+        tmp_chars.extend(c for c in params['chars'].replace(',,', ',').split(',') if c)
+        params['chars'] = tmp_chars
+    else:
+        # Default chars for password
+        params['chars'] = ['ascii_letters', 'digits', ".,:-_"]
+
+    return relpath, params
+
+
 class LookupModule(LookupBase):
 
     def random_password(self, length=DEFAULT_LENGTH, chars=C.DEFAULT_PASSWORD_CHARS):
@@ -62,36 +110,7 @@ class LookupModule(LookupBase):
         ret = []
 
         for term in terms:
-            params = parse_kv(term)
-            if '_raw_params' in params:
-                relpath = params['_raw_params']
-                del params['_raw_params']
-            else:
-                relpath = params
-
-            # Check that we parsed the params correctly
-            if not term.startswith(relpath):
-                # Likely, the user had a non parameter following a parameter.
-                # Reject this as a user typo
-                raise AnsibleError('Unrecognized value after key=value parameters given to password lookup')
-
-            invalid_params = frozenset(params.keys()).difference(VALID_PARAMS)
-            if invalid_params:
-                raise AnsibleError('Unrecognized parameter(s) given to password lookup: %s' % ', '.join(invalid_params))
-
-            length = int(params.get('length', DEFAULT_LENGTH))
-            encrypt = params.get('encrypt', None)
-
-            use_chars = params.get('chars', None)
-            if use_chars:
-                tmp_chars = []
-                if ',,' in use_chars:
-                    tmp_chars.append(',')
-                tmp_chars.extend(use_chars.replace(',,', ',').split(','))
-                use_chars = tmp_chars
-            else:
-                # Default chars for password
-                use_chars = ['ascii_letters', 'digits', ".,:-_"]
+            relpath, params = _parse_parameters(term)
 
             # get password or create it if file doesn't exist
             path = self._loader.path_dwim(relpath)
@@ -102,10 +121,10 @@ class LookupModule(LookupBase):
                 except OSError as e:
                     raise AnsibleError("cannot create the path for the password lookup: %s (error was %s)" % (pathdir, str(e)))
 
-                chars = "".join(getattr(string, c, c) for c in use_chars).replace('"', '').replace("'", '')
-                password = ''.join(random.choice(chars) for _ in range(length))
+                chars = "".join(getattr(string, c, c) for c in params['chars']).replace('"', '').replace("'", '')
+                password = ''.join(random.choice(chars) for _ in range(params['length']))
 
-                if encrypt is not None:
+                if params['encrypt'] is not None:
                     salt = self.random_salt()
                     content = '%s salt=%s' % (password, salt)
                 else:
@@ -125,20 +144,20 @@ class LookupModule(LookupBase):
                     salt = None
 
                 # crypt requested, add salt if missing
-                if (encrypt is not None and not salt):
+                if (params['encrypt'] is not None and not salt):
                     salt = self.random_salt()
                     content = '%s salt=%s' % (password, salt)
                     with open(path, 'w') as f:
                         os.chmod(path, 0o600)
                         f.write(content + '\n')
                 # crypt not requested, remove salt if present
-                elif (encrypt is None and salt):
+                elif (params['encrypt'] is None and salt):
                     with open(path, 'w') as f:
                         os.chmod(path, 0o600)
                         f.write(password + '\n')
 
-            if encrypt:
-                password = do_encrypt(password, encrypt, salt=salt)
+            if params['encrypt']:
+                password = do_encrypt(password, params['encrypt'], salt=salt)
 
             ret.append(password)
 

test/units/plugins/lookup/test_password.py

@@ -0,0 +1,136 @@
+# -*- coding: utf-8 -*-
+# (c) 2015, Toshio Kuratomi <tkuratomi@ansible.com>
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from ansible.compat.tests import unittest
+
+from ansible.plugins.lookup.password import LookupModule, _parse_parameters, DEFAULT_LENGTH
+
+DEFAULT_CHARS = sorted([u'ascii_letters', u'digits', u".,:-_"])
+
+class TestPasswordLookup(unittest.TestCase):
+
+    # Currently there isn't a new-style
+    old_style_params_data = (
+            # Simple case
+            dict(term=u'/path/to/file',
+                filename=u'/path/to/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=DEFAULT_CHARS)
+                ),
+
+            # Special characters in path
+            dict(term=u'/path/with/embedded spaces and/file',
+                filename=u'/path/with/embedded spaces and/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=DEFAULT_CHARS)
+                ),
+            dict(term=u'/path/with/equals/cn=com.ansible',
+                filename=u'/path/with/equals/cn=com.ansible',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=DEFAULT_CHARS)
+                ),
+            dict(term=u'/path/with/unicode/くらとみ/file',
+                filename=u'/path/with/unicode/くらとみ/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=DEFAULT_CHARS)
+                ),
+            # Mix several special chars
+            dict(term=u'/path/with/utf 8 and spaces/くらとみ/file',
+                filename=u'/path/with/utf 8 and spaces/くらとみ/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=DEFAULT_CHARS)
+                ),
+            dict(term=u'/path/with/encoding=unicode/くらとみ/file',
+                filename=u'/path/with/encoding=unicode/くらとみ/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=DEFAULT_CHARS)
+                ),
+            dict(term=u'/path/with/encoding=unicode/くらとみ/and spaces file',
+                filename=u'/path/with/encoding=unicode/くらとみ/and spaces file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=DEFAULT_CHARS)
+                ),
+
+            # Simple parameters
+            dict(term=u'/path/to/file length=42',
+                filename=u'/path/to/file',
+                params=dict(length=42, encrypt=None, chars=DEFAULT_CHARS)
+                ),
+            dict(term=u'/path/to/file encrypt=pbkdf2_sha256',
+                filename=u'/path/to/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt='pbkdf2_sha256', chars=DEFAULT_CHARS)
+                ),
+            dict(term=u'/path/to/file chars=abcdefghijklmnop',
+                filename=u'/path/to/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=[u'abcdefghijklmnop'])
+                ),
+            dict(term=u'/path/to/file chars=digits,abc,def',
+                filename=u'/path/to/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=sorted([u'digits', u'abc', u'def']))
+                ),
+            # Including comma in chars
+            dict(term=u'/path/to/file chars=abcdefghijklmnop,,digits',
+                filename=u'/path/to/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=sorted([u'abcdefghijklmnop', u',', u'digits']))
+                ),
+            dict(term=u'/path/to/file chars=,,',
+                filename=u'/path/to/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=[u','])
+                ),
+
+            # Including = in chars
+            dict(term=u'/path/to/file chars=digits,=,,',
+                filename=u'/path/to/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=sorted([u'digits', u'=', u',']))
+                ),
+            dict(term=u'/path/to/file chars=digits,abc=def',
+                filename=u'/path/to/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=sorted([u'digits', u'abc=def']))
+                ),
+
+            # Including unicode in chars
+            dict(term=u'/path/to/file chars=digits,くらとみ,,',
+                filename=u'/path/to/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=sorted([u'digits', u'くらとみ', u',']))
+                ),
+
+            # Including special chars in both path and chars
+            # Special characters in path
+            dict(term=u'/path/with/embedded spaces and/file chars=abc=def',
+                filename=u'/path/with/embedded spaces and/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=[u'abc=def'])
+                ),
+            dict(term=u'/path/with/equals/cn=com.ansible chars=abc=def',
+                filename=u'/path/with/equals/cn=com.ansible',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=[u'abc=def'])
+                ),
+            dict(term=u'/path/with/unicode/くらとみ/file chars=くらとみ',
+                filename=u'/path/with/unicode/くらとみ/file',
+                params=dict(length=DEFAULT_LENGTH, encrypt=None, chars=[u'くらとみ'])
+                ),
+            )
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    def test_parse_parameters(self):
+        for testcase in self.old_style_params_data:
+            filename, params = _parse_parameters(testcase['term'])
+            params['chars'].sort()
+            self.assertEqual(filename, testcase['filename'])
+            self.assertEqual(params, testcase['params'])