Merge 24c3a20b5f into 6aa8192131

changelogs/fragments/82721-vault-empty.yml

@@ -0,0 +1,4 @@
+---
+bugfixes:
+  - passing a directory as vault password file now raises a meaningful error (https://github.com/ansible/ansible/pull/82721).
+  - empty vault filenames are now silently ignored (https://github.com/ansible/ansible/pull/82721).

lib/ansible/cli/__init__.py

@@ -284,6 +284,10 @@ class CLI(ABC):
                 loader.set_vault_secrets(vault_secrets)
                 continue
 
+            if not vault_id_value:
+                # silently ignore empty filenames
+                continue
+
             # assuming anything else is a password file
             display.vvvvv('Reading vault password file: %s' % vault_id_value)
             # read vault_pass from a file

lib/ansible/parsing/vault/__init__.py

@@ -356,6 +356,8 @@ def get_file_vault_secret(filename=None, vault_id=None, encoding=None, loader=No
     this_path = unfrackpath(filename, follow=False)
     if not os.path.exists(this_path):
         raise AnsibleError("The vault password file %s was not found" % this_path)
+    if not os.path.isfile(this_path):
+        raise AnsibleError("The vault password file %s is not a file" % this_path)
 
     # it is a script?
     if loader.is_executable(this_path):

test/units/cli/test_cli.py

@@ -22,6 +22,7 @@ from unittest.mock import patch, MagicMock
 
 from units.mock.loader import DictDataLoader
 
+from ansible.errors import AnsibleError
 from ansible.release import __version__
 from ansible.parsing import vault
 from ansible import cli
@@ -357,3 +358,22 @@ class TestCliSetupVaultSecrets(unittest.TestCase):
         self.assertIsInstance(res, list)
         match = vault.match_secrets(res, ['some_vault_id'])[0][1]
         self.assertEqual(match.bytes, b'prompt1_password')
+
+    def test_empty_slug(self):
+        res = cli.CLI.setup_vault_secrets(loader=self.fake_loader,
+                                          vault_ids=[''])
+        self.assertIsInstance(res, list)
+        self.assertEqual(0, len(res))
+
+    def test_empty_name_part(self):
+        self.assertRaisesRegex(AnsibleError,
+                               '.*The vault password file .*/foo was not found.*',
+                               cli.CLI.setup_vault_secrets,
+                               loader=self.fake_loader,
+                               vault_ids=['@foo'])
+
+    def test_empty_value_part(self):
+        res = cli.CLI.setup_vault_secrets(loader=self.fake_loader,
+                                          vault_ids=['foo@'])
+        self.assertIsInstance(res, list)
+        self.assertEqual(0, len(res))

test/units/parsing/vault/test_vault.py

@@ -358,6 +358,9 @@ class TestScriptIsClient(unittest.TestCase):
 
 
 class TestGetFileVaultSecret(unittest.TestCase):
+    def setUp(self):
+        self.fake_loader = DictDataLoader({})
+
     def test_file(self):
         password = 'some password'
 
@@ -365,37 +368,67 @@ class TestGetFileVaultSecret(unittest.TestCase):
         tmp_file.write(to_bytes(password))
         tmp_file.close()
 
-        fake_loader = DictDataLoader({tmp_file.name: 'sdfadf'})
-
-        secret = vault.get_file_vault_secret(filename=tmp_file.name, loader=fake_loader)
+        secret = vault.get_file_vault_secret(filename=tmp_file.name, loader=self.fake_loader)
         secret.load()
 
         os.unlink(tmp_file.name)
 
         self.assertEqual(secret.bytes, to_bytes(password))
 
-    def test_file_not_a_directory(self):
+    def test_file_path_part_not_a_directory(self):
         filename = '/dev/null/foobar'
-        fake_loader = DictDataLoader({filename: 'sdfadf'})
 
         self.assertRaisesRegex(errors.AnsibleError,
                                '.*The vault password file %s was not found.*' % filename,
                                vault.get_file_vault_secret,
-                               filename=filename,
-                               loader=fake_loader)
+                               filename=filename)
 
     def test_file_not_found(self):
-        tmp_file = tempfile.NamedTemporaryFile()
-        filename = os.path.realpath(tmp_file.name)
-        tmp_file.close()
-
-        fake_loader = DictDataLoader({filename: 'sdfadf'})
+        with tempfile.NamedTemporaryFile() as tmp_file:
+            filename = os.path.realpath(tmp_file.name)
 
         self.assertRaisesRegex(errors.AnsibleError,
                                '.*The vault password file %s was not found.*' % filename,
                                vault.get_file_vault_secret,
-                               filename=filename,
-                               loader=fake_loader)
+                               filename=filename)
+
+    def test_file_not_a_directory(self):
+        filename = '/dev'
+
+        self.assertRaisesRegex(errors.AnsibleError,
+                               '.*The vault password file %s is not a file.*' % filename,
+                               vault.get_file_vault_secret,
+                               filename=filename)
+
+    def test_file_is_symlink_to_file(self):
+        with tempfile.NamedTemporaryFile(suffix='symlink') as tmp_file:
+            symlink_filename = os.path.realpath(tmp_file.name)
+
+        try:
+            with tempfile.NamedTemporaryFile(suffix='target') as tmp_file:
+                target_filename = tmp_file.name
+                os.symlink(target_filename, symlink_filename)
+
+                secret = vault.get_file_vault_secret(filename=symlink_filename, loader=self.fake_loader)
+                self.assertIsInstance(secret, vault.FileVaultSecret)
+                self.assertEqual(secret.filename, symlink_filename)
+        finally:
+            os.remove(symlink_filename)
+
+    def test_file_is_symlink_to_directory(self):
+        with tempfile.NamedTemporaryFile(suffix='symlink') as tmp_file:
+            symlink_filename = os.path.realpath(tmp_file.name)
+
+        try:
+            with tempfile.TemporaryDirectory(suffix='target') as target_filename:
+                os.symlink(target_filename, symlink_filename)
+
+                self.assertRaisesRegex(errors.AnsibleError,
+                                       '.*The vault password file %s is not a file.*' % symlink_filename,
+                                       vault.get_file_vault_secret,
+                                       filename=symlink_filename)
+        finally:
+            os.remove(symlink_filename)
 
 
 class TestVaultIsEncrypted(unittest.TestCase):