You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
ansible/test/units/parsing/test_dataloader.py

240 lines
9.7 KiB
Python

# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.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
import os
from units.compat import unittest
from unittest.mock import patch, mock_open
from ansible.errors import AnsibleParserError, yaml_strings, AnsibleFileNotFound
from ansible.parsing.vault import AnsibleVaultError
from ansible.module_utils._text import to_text
from ansible.module_utils.six import PY3
Support multiple vault passwords (#22756) Fixes #13243 ** Add --vault-id to name/identify multiple vault passwords Use --vault-id to indicate id and path/type --vault-id=prompt # prompt for default vault id password --vault-id=myorg@prompt # prompt for a vault_id named 'myorg' --vault-id=a_password_file # load ./a_password_file for default id --vault-id=myorg@a_password_file # load file for 'myorg' vault id vault_id's are created implicitly for existing --vault-password-file and --ask-vault-pass options. Vault ids are just for UX purposes and bookkeeping. Only the vault payload and the password bytestring is needed to decrypt a vault blob. Replace passing password around everywhere with a VaultSecrets object. If we specify a vault_id, mention that in password prompts Specifying multiple -vault-password-files will now try each until one works ** Rev vault format in a backwards compatible way The 1.2 vault format adds the vault_id to the header line of the vault text. This is backwards compatible with older versions of ansible. Old versions will just ignore it and treat it as the default (and only) vault id. Note: only 2.4+ supports multiple vault passwords, so while earlier ansible versions can read the vault-1.2 format, it does not make them magically support multiple vault passwords. use 1.1 format for 'default' vault_id Vaulted items that need to include a vault_id will be written in 1.2 format. If we set a new DEFAULT_VAULT_IDENTITY, then the default will use version 1.2 vault will only use a vault_id if one is specified. So if none is specified and C.DEFAULT_VAULT_IDENTITY is 'default' we use the old format. ** Changes/refactors needed to implement multiple vault passwords raise exceptions on decrypt fail, check vault id early split out parsing the vault plaintext envelope (with the sha/original plaintext) to _split_plaintext_envelope() some cli fixups for specifying multiple paths in the unfrack_paths optparse callback fix py3 dict.keys() 'dict_keys object is not indexable' error pluralize cli.options.vault_password_file -> vault_password_files pluralize cli.options.new_vault_password_file -> new_vault_password_files pluralize cli.options.vault_id -> cli.options.vault_ids ** Add a config option (vault_id_match) to force vault id matching. With 'vault_id_match=True' and an ansible vault that provides a vault_id, then decryption will require that a matching vault_id is required. (via --vault-id=my_vault_id@password_file, for ex). In other words, if the config option is true, then only the vault secrets with matching vault ids are candidates for decrypting a vault. If option is false (the default), then all of the provided vault secrets will be selected. If a user doesn't want all vault secrets to be tried to decrypt any vault content, they can enable this option. Note: The vault id used for the match is not encrypted or cryptographically signed. It is just a label/id/nickname used for referencing a specific vault secret.
7 years ago
from units.mock.vault_helper import TextVaultSecret
from ansible.parsing.dataloader import DataLoader
from units.mock.path import mock_unfrackpath_noop
class TestDataLoader(unittest.TestCase):
def setUp(self):
self._loader = DataLoader()
@patch('os.path.exists')
def test__is_role(self, p_exists):
p_exists.side_effect = lambda p: p == b'test_path/tasks/main.yml'
self.assertTrue(self._loader._is_role('test_path/tasks'))
self.assertTrue(self._loader._is_role('test_path/'))
@patch.object(DataLoader, '_get_file_contents')
def test_parse_json_from_file(self, mock_def):
mock_def.return_value = (b"""{"a": 1, "b": 2, "c": 3}""", True)
output = self._loader.load_from_file('dummy_json.txt')
self.assertEqual(output, dict(a=1, b=2, c=3))
@patch.object(DataLoader, '_get_file_contents')
def test_parse_yaml_from_file(self, mock_def):
mock_def.return_value = (b"""
a: 1
b: 2
c: 3
""", True)
output = self._loader.load_from_file('dummy_yaml.txt')
self.assertEqual(output, dict(a=1, b=2, c=3))
@patch.object(DataLoader, '_get_file_contents')
def test_parse_fail_from_file(self, mock_def):
mock_def.return_value = (b"""
TEXT:
***
NOT VALID
""", True)
self.assertRaises(AnsibleParserError, self._loader.load_from_file, 'dummy_yaml_bad.txt')
@patch('ansible.errors.AnsibleError._get_error_lines_from_file')
@patch.object(DataLoader, '_get_file_contents')
def test_tab_error(self, mock_def, mock_get_error_lines):
mock_def.return_value = (u"""---\nhosts: localhost\nvars:\n foo: bar\n\tblip: baz""", True)
mock_get_error_lines.return_value = ('''\tblip: baz''', '''..foo: bar''')
with self.assertRaises(AnsibleParserError) as cm:
self._loader.load_from_file('dummy_yaml_text.txt')
self.assertIn(yaml_strings.YAML_COMMON_LEADING_TAB_ERROR, str(cm.exception))
self.assertIn('foo: bar', str(cm.exception))
@patch('ansible.parsing.dataloader.unfrackpath', mock_unfrackpath_noop)
@patch.object(DataLoader, '_is_role')
def test_path_dwim_relative(self, mock_is_role):
"""
simulate a nested dynamic include:
playbook.yml:
- hosts: localhost
roles:
- { role: 'testrole' }
testrole/tasks/main.yml:
- include: "include1.yml"
static: no
testrole/tasks/include1.yml:
- include: include2.yml
static: no
testrole/tasks/include2.yml:
- debug: msg="blah"
"""
mock_is_role.return_value = False
with patch('os.path.exists') as mock_os_path_exists:
mock_os_path_exists.return_value = False
self._loader.path_dwim_relative('/tmp/roles/testrole/tasks', 'tasks', 'included2.yml')
# Fetch first args for every call
# mock_os_path_exists.assert_any_call isn't used because os.path.normpath must be used in order to compare paths
called_args = [os.path.normpath(to_text(call[0][0])) for call in mock_os_path_exists.call_args_list]
# 'path_dwim_relative' docstrings say 'with or without explicitly named dirname subdirs':
self.assertIn('/tmp/roles/testrole/tasks/included2.yml', called_args)
self.assertIn('/tmp/roles/testrole/tasks/tasks/included2.yml', called_args)
# relative directories below are taken in account too:
self.assertIn('tasks/included2.yml', called_args)
self.assertIn('included2.yml', called_args)
def test_path_dwim_root(self):
self.assertEqual(self._loader.path_dwim('/'), '/')
def test_path_dwim_home(self):
self.assertEqual(self._loader.path_dwim('~'), os.path.expanduser('~'))
def test_path_dwim_tilde_slash(self):
self.assertEqual(self._loader.path_dwim('~/'), os.path.expanduser('~'))
def test_get_real_file(self):
self.assertEqual(self._loader.get_real_file(__file__), __file__)
def test_is_file(self):
self.assertTrue(self._loader.is_file(__file__))
def test_is_directory_positive(self):
self.assertTrue(self._loader.is_directory(os.path.dirname(__file__)))
def test_get_file_contents_none_path(self):
self.assertRaisesRegex(AnsibleParserError, 'Invalid filename',
self._loader._get_file_contents, None)
def test_get_file_contents_non_existent_path(self):
self.assertRaises(AnsibleFileNotFound, self._loader._get_file_contents, '/non_existent_file')
class TestPathDwimRelativeDataLoader(unittest.TestCase):
def setUp(self):
self._loader = DataLoader()
def test_all_slash(self):
self.assertEqual(self._loader.path_dwim_relative('/', '/', '/'), '/')
def test_path_endswith_role(self):
self.assertEqual(self._loader.path_dwim_relative(path='foo/bar/tasks/', dirname='/', source='/'), '/')
def test_path_endswith_role_main_yml(self):
self.assertIn('main.yml', self._loader.path_dwim_relative(path='foo/bar/tasks/', dirname='/', source='main.yml'))
def test_path_endswith_role_source_tilde(self):
self.assertEqual(self._loader.path_dwim_relative(path='foo/bar/tasks/', dirname='/', source='~/'), os.path.expanduser('~'))
class TestPathDwimRelativeStackDataLoader(unittest.TestCase):
def setUp(self):
self._loader = DataLoader()
def test_none(self):
self.assertRaisesRegex(AnsibleFileNotFound, 'on the Ansible Controller', self._loader.path_dwim_relative_stack, None, None, None)
def test_empty_strings(self):
self.assertEqual(self._loader.path_dwim_relative_stack('', '', ''), './')
def test_empty_lists(self):
self.assertEqual(self._loader.path_dwim_relative_stack([], '', '~/'), os.path.expanduser('~'))
def test_all_slash(self):
self.assertEqual(self._loader.path_dwim_relative_stack('/', '/', '/'), '/')
def test_path_endswith_role(self):
self.assertEqual(self._loader.path_dwim_relative_stack(paths=['foo/bar/tasks/'], dirname='/', source='/'), '/')
def test_path_endswith_role_source_tilde(self):
self.assertEqual(self._loader.path_dwim_relative_stack(paths=['foo/bar/tasks/'], dirname='/', source='~/'), os.path.expanduser('~'))
def test_path_endswith_role_source_main_yml(self):
self.assertRaises(AnsibleFileNotFound, self._loader.path_dwim_relative_stack, ['foo/bar/tasks/'], '/', 'main.yml')
def test_path_endswith_role_source_main_yml_source_in_dirname(self):
self.assertRaises(AnsibleFileNotFound, self._loader.path_dwim_relative_stack, 'foo/bar/tasks/', 'tasks', 'tasks/main.yml')
class TestDataLoaderWithVault(unittest.TestCase):
def setUp(self):
self._loader = DataLoader()
Support multiple vault passwords (#22756) Fixes #13243 ** Add --vault-id to name/identify multiple vault passwords Use --vault-id to indicate id and path/type --vault-id=prompt # prompt for default vault id password --vault-id=myorg@prompt # prompt for a vault_id named 'myorg' --vault-id=a_password_file # load ./a_password_file for default id --vault-id=myorg@a_password_file # load file for 'myorg' vault id vault_id's are created implicitly for existing --vault-password-file and --ask-vault-pass options. Vault ids are just for UX purposes and bookkeeping. Only the vault payload and the password bytestring is needed to decrypt a vault blob. Replace passing password around everywhere with a VaultSecrets object. If we specify a vault_id, mention that in password prompts Specifying multiple -vault-password-files will now try each until one works ** Rev vault format in a backwards compatible way The 1.2 vault format adds the vault_id to the header line of the vault text. This is backwards compatible with older versions of ansible. Old versions will just ignore it and treat it as the default (and only) vault id. Note: only 2.4+ supports multiple vault passwords, so while earlier ansible versions can read the vault-1.2 format, it does not make them magically support multiple vault passwords. use 1.1 format for 'default' vault_id Vaulted items that need to include a vault_id will be written in 1.2 format. If we set a new DEFAULT_VAULT_IDENTITY, then the default will use version 1.2 vault will only use a vault_id if one is specified. So if none is specified and C.DEFAULT_VAULT_IDENTITY is 'default' we use the old format. ** Changes/refactors needed to implement multiple vault passwords raise exceptions on decrypt fail, check vault id early split out parsing the vault plaintext envelope (with the sha/original plaintext) to _split_plaintext_envelope() some cli fixups for specifying multiple paths in the unfrack_paths optparse callback fix py3 dict.keys() 'dict_keys object is not indexable' error pluralize cli.options.vault_password_file -> vault_password_files pluralize cli.options.new_vault_password_file -> new_vault_password_files pluralize cli.options.vault_id -> cli.options.vault_ids ** Add a config option (vault_id_match) to force vault id matching. With 'vault_id_match=True' and an ansible vault that provides a vault_id, then decryption will require that a matching vault_id is required. (via --vault-id=my_vault_id@password_file, for ex). In other words, if the config option is true, then only the vault secrets with matching vault ids are candidates for decrypting a vault. If option is false (the default), then all of the provided vault secrets will be selected. If a user doesn't want all vault secrets to be tried to decrypt any vault content, they can enable this option. Note: The vault id used for the match is not encrypted or cryptographically signed. It is just a label/id/nickname used for referencing a specific vault secret.
7 years ago
vault_secrets = [('default', TextVaultSecret('ansible'))]
self._loader.set_vault_secrets(vault_secrets)
self.test_vault_data_path = os.path.join(os.path.dirname(__file__), 'fixtures', 'vault.yml')
def tearDown(self):
pass
def test_get_real_file_vault(self):
real_file_path = self._loader.get_real_file(self.test_vault_data_path)
self.assertTrue(os.path.exists(real_file_path))
def test_get_real_file_vault_no_vault(self):
self._loader.set_vault_secrets(None)
self.assertRaises(AnsibleParserError, self._loader.get_real_file, self.test_vault_data_path)
def test_get_real_file_vault_wrong_password(self):
wrong_vault = [('default', TextVaultSecret('wrong_password'))]
self._loader.set_vault_secrets(wrong_vault)
self.assertRaises(AnsibleVaultError, self._loader.get_real_file, self.test_vault_data_path)
def test_get_real_file_not_a_path(self):
self.assertRaisesRegex(AnsibleParserError, 'Invalid filename', self._loader.get_real_file, None)
@patch.multiple(DataLoader, path_exists=lambda s, x: True, is_file=lambda s, x: True)
def test_parse_from_vault_1_1_file(self):
vaulted_data = """$ANSIBLE_VAULT;1.1;AES256
33343734386261666161626433386662623039356366656637303939306563376130623138626165
6436333766346533353463636566313332623130383662340a393835656134633665333861393331
37666233346464636263636530626332623035633135363732623332313534306438393366323966
3135306561356164310a343937653834643433343734653137383339323330626437313562306630
3035
"""
if PY3:
builtins_name = 'builtins'
else:
builtins_name = '__builtin__'
with patch(builtins_name + '.open', mock_open(read_data=vaulted_data.encode('utf-8'))):
output = self._loader.load_from_file('dummy_vault.txt')
self.assertEqual(output, dict(foo='bar'))