Vault secrets empty password (#28186)
* Better handling of empty/invalid passwords empty password files are global error and cause an exit. A warning is also emitted with more detail. ie, if any of the password/secret sources provide a bogus password (ie, empty) or fail (exception, ctrl-d, EOFError), we stop at the first error and exit. This makes behavior when entering empty password at prompt match 2.3 (ie, an error)
This commit is contained in:
Adrian Likins
GitHub
parent
271127113f
commit
e287af1ac8
5 changed files with 111 additions and 20 deletions
|
|
@ -34,7 +34,7 @@ from abc import ABCMeta, abstractmethod
|
|||
|
||||
import ansible
|
||||
from ansible import constants as C
|
||||
from ansible.errors import AnsibleOptionsError
|
||||
from ansible.errors import AnsibleOptionsError, AnsibleError
|
||||
from ansible.inventory.manager import InventoryManager
|
||||
from ansible.module_utils.six import with_metaclass, string_types
|
||||
from ansible.module_utils._text import to_bytes, to_text
|
||||
|
|
@ -180,7 +180,8 @@ class CLI(with_metaclass(ABCMeta, object)):
|
|||
return ret
|
||||
|
||||
@staticmethod
|
||||
def build_vault_ids(vault_ids, vault_password_files=None, ask_vault_pass=None):
|
||||
def build_vault_ids(vault_ids, vault_password_files=None,
|
||||
ask_vault_pass=None, create_new_password=None):
|
||||
vault_password_files = vault_password_files or []
|
||||
vault_ids = vault_ids or []
|
||||
|
||||
|
|
@ -193,7 +194,9 @@ class CLI(with_metaclass(ABCMeta, object)):
|
|||
# used by --vault-id and --vault-password-file
|
||||
vault_ids.append(id_slug)
|
||||
|
||||
if ask_vault_pass:
|
||||
# if an action needs an encrypt password (create_new_password=True) and we dont
|
||||
# have other secrets setup, then automatically add a password prompt as well.
|
||||
if ask_vault_pass or (create_new_password and not vault_ids):
|
||||
id_slug = u'%s@%s' % (C.DEFAULT_VAULT_IDENTITY, u'prompt_ask_vault_pass')
|
||||
vault_ids.append(id_slug)
|
||||
|
||||
|
|
@ -233,21 +236,26 @@ class CLI(with_metaclass(ABCMeta, object)):
|
|||
for vault_id_slug in vault_ids:
|
||||
vault_id_name, vault_id_value = CLI.split_vault_id(vault_id_slug)
|
||||
if vault_id_value in ['prompt', 'prompt_ask_vault_pass']:
|
||||
|
||||
# --vault-id some_name@prompt_ask_vault_pass --vault-id other_name@prompt_ask_vault_pass will be a little
|
||||
# confusing since it will use the old format without the vault id in the prompt
|
||||
if vault_id_name:
|
||||
prompted_vault_secret = PromptVaultSecret(prompt_formats=prompt_formats[vault_id_value],
|
||||
vault_id=vault_id_name)
|
||||
built_vault_id = vault_id_name or C.DEFAULT_VAULT_IDENTITY
|
||||
|
||||
# choose the prompt based on --vault-id=prompt or --ask-vault-pass. --ask-vault-pass
|
||||
# always gets the old format for Tower compatibility.
|
||||
# ie, we used --ask-vault-pass, so we need to use the old vault password prompt
|
||||
# format since Tower needs to match on that format.
|
||||
prompted_vault_secret = PromptVaultSecret(prompt_formats=prompt_formats[vault_id_value],
|
||||
vault_id=built_vault_id)
|
||||
|
||||
# a empty or invalid password from the prompt will warn and continue to the next
|
||||
# without erroring globablly
|
||||
try:
|
||||
prompted_vault_secret.load()
|
||||
vault_secrets.append((vault_id_name, prompted_vault_secret))
|
||||
else:
|
||||
# ie, we used --ask-vault-pass, so we need to use the old vault password prompt
|
||||
# format since Tower needs to match on that format.
|
||||
prompted_vault_secret = PromptVaultSecret(prompt_formats=prompt_formats[vault_id_value],
|
||||
vault_id=C.DEFAULT_VAULT_IDENTITY)
|
||||
prompted_vault_secret.load()
|
||||
vault_secrets.append((C.DEFAULT_VAULT_IDENTITY, prompted_vault_secret))
|
||||
except AnsibleError as exc:
|
||||
display.warning('Error in vault password prompt (%s): %s' % (vault_id_name, exc))
|
||||
raise
|
||||
|
||||
vault_secrets.append((built_vault_id, prompted_vault_secret))
|
||||
|
||||
# update loader with new secrets incrementally, so we can load a vault password
|
||||
# that is encrypted with a vault secret provided earlier
|
||||
|
|
@ -260,7 +268,14 @@ class CLI(with_metaclass(ABCMeta, object)):
|
|||
file_vault_secret = get_file_vault_secret(filename=vault_id_value,
|
||||
vault_id_name=vault_id_name,
|
||||
loader=loader)
|
||||
file_vault_secret.load()
|
||||
|
||||
# an invalid password file will error globally
|
||||
try:
|
||||
file_vault_secret.load()
|
||||
except AnsibleError as exc:
|
||||
display.warning('Error in vault password file loading (%s): %s' % (vault_id_name, exc))
|
||||
raise
|
||||
|
||||
if vault_id_name:
|
||||
vault_secrets.append((vault_id_name, file_vault_secret))
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -101,6 +101,10 @@ class AnsibleVaultError(AnsibleError):
|
|||
pass
|
||||
|
||||
|
||||
class AnsibleVaultPasswordError(AnsibleVaultError):
|
||||
pass
|
||||
|
||||
|
||||
def is_encrypted(data):
|
||||
""" Test if this is vault encrypted data blob
|
||||
|
||||
|
|
@ -218,6 +222,18 @@ def format_vaulttext_envelope(b_ciphertext, cipher_name, version=None, vault_id=
|
|||
return b_vaulttext
|
||||
|
||||
|
||||
def verify_secret_is_not_empty(secret, msg=None):
|
||||
'''Check the secret against minimal requirements.
|
||||
|
||||
Raises: AnsibleVaultPasswordError if the password does not meet requirements.
|
||||
|
||||
Currently, only requirement is that the password is not None or an empty string.
|
||||
'''
|
||||
msg = msg or 'Invalid vault password was provided'
|
||||
if not secret:
|
||||
raise AnsibleVaultPasswordError(msg)
|
||||
|
||||
|
||||
class VaultSecret:
|
||||
'''Opaque/abstract objects for a single vault secret. ie, a password or a key.'''
|
||||
def __init__(self, _bytes=None):
|
||||
|
|
@ -263,7 +279,10 @@ class PromptVaultSecret(VaultSecret):
|
|||
try:
|
||||
vault_pass = display.prompt(prompt, private=True)
|
||||
except EOFError:
|
||||
break
|
||||
raise AnsibleVaultError('EOFError (ctrl-d) on prompt for (%s)' % self.vault_id)
|
||||
|
||||
verify_secret_is_not_empty(vault_pass)
|
||||
|
||||
b_vault_pass = to_bytes(vault_pass, errors='strict', nonstring='simplerepr').strip()
|
||||
b_vault_passwords.append(b_vault_pass)
|
||||
|
||||
|
|
@ -335,6 +354,9 @@ class FileVaultSecret(VaultSecret):
|
|||
except (OSError, IOError) as e:
|
||||
raise AnsibleError("Could not read vault password file %s: %s" % (filename, e))
|
||||
|
||||
verify_secret_is_not_empty(vault_pass,
|
||||
msg='Invalid vault password was provided from file (%s)' % filename)
|
||||
|
||||
return vault_pass
|
||||
|
||||
def __repr__(self):
|
||||
|
|
@ -364,6 +386,8 @@ class ScriptVaultSecret(FileVaultSecret):
|
|||
raise AnsibleError("Vault password script %s returned non-zero (%s): %s" % (filename, p.returncode, stderr))
|
||||
|
||||
vault_pass = stdout.strip(b'\r\n')
|
||||
verify_secret_is_not_empty(vault_pass,
|
||||
msg='Invalid vault password was provided from script (%s)' % filename)
|
||||
return vault_pass
|
||||
|
||||
|
||||
|
|
|
|||
0
test/integration/targets/vault/empty-password
Normal file
0
test/integration/targets/vault/empty-password
Normal file
|
|
@ -14,6 +14,7 @@ echo "This is a test file for format 1.2" > "${TEST_FILE_1_2}"
|
|||
|
||||
TEST_FILE_OUTPUT="${MYTMPDIR}/test_file_output"
|
||||
|
||||
|
||||
# old format
|
||||
ansible-vault view "$@" --vault-password-file vault-password-ansible format_1_0_AES.yml
|
||||
|
||||
|
|
@ -38,6 +39,7 @@ echo "rc was $WRONG_RC (1 is expected)"
|
|||
|
||||
set -eux
|
||||
|
||||
|
||||
# new format, view
|
||||
ansible-vault view "$@" --vault-password-file vault-password format_1_1_AES256.yml
|
||||
|
||||
|
|
@ -184,6 +186,24 @@ ansible-vault encrypt "$@" --vault-password-file "${NEW_VAULT_PASSWORD}" --outpu
|
|||
ansible-vault view "$@" --vault-password-file "${NEW_VAULT_PASSWORD}" - < "${TEST_FILE_OUTPUT}"
|
||||
ansible-vault decrypt "$@" --vault-password-file "${NEW_VAULT_PASSWORD}" --output=- < "${TEST_FILE_OUTPUT}"
|
||||
|
||||
# test using an empty vault password file
|
||||
ansible-vault view "$@" --vault-password-file empty-password format_1_1_AES256.yml && :
|
||||
WRONG_RC=$?
|
||||
echo "rc was $WRONG_RC (1 is expected)"
|
||||
[ $WRONG_RC -eq 1 ]
|
||||
|
||||
ansible-vault view "$@" --vault-id=empty@empty-password --vault-password-file empty-password format_1_1_AES256.yml && :
|
||||
WRONG_RC=$?
|
||||
echo "rc was $WRONG_RC (1 is expected)"
|
||||
[ $WRONG_RC -eq 1 ]
|
||||
|
||||
echo 'foo' > some_file.txt
|
||||
ansible-vault encrypt "$@" --vault-password-file empty-password some_file.txt && :
|
||||
WRONG_RC=$?
|
||||
echo "rc was $WRONG_RC (1 is expected)"
|
||||
[ $WRONG_RC -eq 1 ]
|
||||
|
||||
|
||||
ansible-vault encrypt_string "$@" --vault-password-file "${NEW_VAULT_PASSWORD}" "a test string"
|
||||
|
||||
ansible-vault encrypt_string "$@" --vault-password-file "${NEW_VAULT_PASSWORD}" --name "blippy" "a test string names blippy"
|
||||
|
|
@ -280,3 +300,9 @@ WRONG_RC=$?
|
|||
echo "rc was $WRONG_RC (1 is expected)"
|
||||
[ $WRONG_RC -eq 1 ]
|
||||
|
||||
# with empty password file
|
||||
ansible-playbook test_vault.yml -i ../../inventory -v "$@" --vault-id empty@empty-password && :
|
||||
WRONG_RC=$?
|
||||
echo "rc was $WRONG_RC (1 is expected)"
|
||||
[ $WRONG_RC -eq 1 ]
|
||||
|
||||
|
|
|
|||
|
|
@ -77,8 +77,9 @@ class TestPromptVaultSecret(unittest.TestCase):
|
|||
@patch('ansible.parsing.vault.display.prompt', side_effect=EOFError)
|
||||
def test_prompt_eoferror(self, mock_display_prompt):
|
||||
secret = vault.PromptVaultSecret(vault_id='test_id')
|
||||
secret.load()
|
||||
self.assertEqual(secret._bytes, None)
|
||||
self.assertRaisesRegexp(vault.AnsibleVaultError,
|
||||
'EOFError.*test_id',
|
||||
secret.load)
|
||||
|
||||
@patch('ansible.parsing.vault.display.prompt', side_effect=['first_password', 'second_password'])
|
||||
def test_prompt_passwords_dont_match(self, mock_display_prompt):
|
||||
|
|
@ -129,6 +130,21 @@ class TestFileVaultSecret(unittest.TestCase):
|
|||
|
||||
self.assertEqual(secret.bytes, to_bytes(password))
|
||||
|
||||
def test_file_empty(self):
|
||||
|
||||
tmp_file = tempfile.NamedTemporaryFile(delete=False)
|
||||
tmp_file.write(to_bytes(''))
|
||||
tmp_file.close()
|
||||
|
||||
fake_loader = DictDataLoader({tmp_file.name: ''})
|
||||
|
||||
secret = vault.FileVaultSecret(loader=fake_loader, filename=tmp_file.name)
|
||||
self.assertRaisesRegexp(vault.AnsibleVaultPasswordError,
|
||||
'Invalid vault password was provided from file.*%s' % tmp_file.name,
|
||||
secret.load)
|
||||
|
||||
os.unlink(tmp_file.name)
|
||||
|
||||
def test_file_not_a_directory(self):
|
||||
filename = '/dev/null/foobar'
|
||||
fake_loader = DictDataLoader({filename: 'sdfadf'})
|
||||
|
|
@ -166,12 +182,22 @@ class TestScriptVaultSecret(unittest.TestCase):
|
|||
|
||||
@patch('ansible.parsing.vault.subprocess.Popen')
|
||||
def test_read_file(self, mock_popen):
|
||||
self._mock_popen(mock_popen)
|
||||
self._mock_popen(mock_popen, stdout=b'some_password')
|
||||
secret = vault.ScriptVaultSecret()
|
||||
with patch.object(secret, 'loader') as mock_loader:
|
||||
mock_loader.is_executable = MagicMock(return_value=True)
|
||||
secret.load()
|
||||
|
||||
@patch('ansible.parsing.vault.subprocess.Popen')
|
||||
def test_read_file_empty(self, mock_popen):
|
||||
self._mock_popen(mock_popen, stdout=b'')
|
||||
secret = vault.ScriptVaultSecret()
|
||||
with patch.object(secret, 'loader') as mock_loader:
|
||||
mock_loader.is_executable = MagicMock(return_value=True)
|
||||
self.assertRaisesRegexp(vault.AnsibleVaultPasswordError,
|
||||
'Invalid vault password was provided from script',
|
||||
secret.load)
|
||||
|
||||
@patch('ansible.parsing.vault.subprocess.Popen')
|
||||
def test_read_file_os_error(self, mock_popen):
|
||||
self._mock_popen(mock_popen)
|
||||
|
|
|
|||
Loading…
Reference in a new issue