Implement vault encrypted yaml variables. (#16274)

Make !vault-encrypted create a AnsibleVaultUnicode
yaml object that can be used as a regular string object.

This allows a playbook to include a encrypted vault
blob for the value of a yaml variable. A 'secret_password'
variable can have it's value encrypted instead of having
to vault encrypt an entire vars file.

Add __ENCRYPTED__ to the vault yaml types so
template.Template can treat it similar
to __UNSAFE__ flags.

vault.VaultLib api changes:
    - Split VaultLib.encrypt to encrypt and encrypt_bytestring

    - VaultLib.encrypt() previously accepted the plaintext data
      as either a byte string or a unicode string.
      Doing the right thing based on the input type would fail
      on py3 if given a arg of type 'bytes'. To simplify the
      API, vaultlib.encrypt() now assumes input plaintext is a
      py2 unicode or py3 str. It will encode to utf-8 then call
      the new encrypt_bytestring(). The new methods are less
      ambiguous.

    - moved VaultLib.is_encrypted logic to vault module scope
      and split to is_encrypted() and is_encrypted_file().

Add a test/unit/mock/yaml_helper.py
It has some helpers for testing parsing/yaml

Integration tests added as roles test_vault and test_vault_embedded
This commit is contained in:
Adrian Likins 2016-08-23 20:03:11 -04:00 committed by GitHub
commit e396d5d508
21 changed files with 934 additions and 111 deletions

View file

@ -20,14 +20,12 @@ from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from six import PY3
from yaml.scanner import ScannerError
from ansible.compat.tests import unittest
from ansible.compat.tests.mock import patch, mock_open
from ansible.errors import AnsibleParserError
from ansible.parsing.dataloader import DataLoader
from ansible.parsing.yaml.objects import AnsibleMapping
class TestDataLoader(unittest.TestCase):
@ -85,6 +83,6 @@ class TestDataLoaderWithVault(unittest.TestCase):
else:
builtins_name = '__builtin__'
with patch(builtins_name + '.open', mock_open(read_data=vaulted_data)):
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'))

View file

@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
#
# This file is part of Ansible
@ -19,14 +20,12 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import getpass
import os
import shutil
import time
import tempfile
import six
from binascii import unhexlify
import binascii
import io
import os
from binascii import hexlify
from nose.plugins.skip import SkipTest
@ -35,6 +34,7 @@ from ansible.utils.unicode import to_bytes, to_unicode
from ansible import errors
from ansible.parsing.vault import VaultLib
from ansible.parsing import vault
# Counter import fails for 2.0.1, requires >= 2.6.1 from pip
try:
@ -57,6 +57,150 @@ try:
except ImportError:
HAS_AES = False
class TestVaultIsEncrypted(unittest.TestCase):
def test_utf8_not_encrypted(self):
b_data = "foobar".encode('utf8')
self.assertFalse(vault.is_encrypted(b_data))
def test_utf8_encrypted(self):
data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
b_data = data.encode('utf8')
self.assertTrue(vault.is_encrypted(b_data))
def test_bytes_not_encrypted(self):
b_data = b"foobar"
self.assertFalse(vault.is_encrypted(b_data))
def test_bytes_encrypted(self):
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" + hexlify(b"ansible")
self.assertTrue(vault.is_encrypted(b_data))
def test_unicode_not_encrypted_py3(self):
if not six.PY3:
raise SkipTest()
data = u"ァ ア ィ イ ゥ ウ ェ エ ォ オ カ ガ キ ギ ク グ ケ "
self.assertRaises(TypeError, vault.is_encrypted, data)
def test_unicode_not_encrypted_py2(self):
if six.PY3:
raise SkipTest()
data = u"ァ ア ィ イ ゥ ウ ェ エ ォ オ カ ガ キ ギ ク グ ケ "
# py2 will take a unicode string, but that should always fails
self.assertFalse(vault.is_encrypted(data))
def test_unicode_is_encrypted_py3(self):
if not six.PY3:
raise SkipTest()
data = "$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
# should still be a type error
self.assertRaises(TypeError, vault.is_encrypted, data)
def test_unicode_is_encrypted_py2(self):
if six.PY3:
raise SkipTest()
data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
# THis works, but arguably shouldn't...
self.assertTrue(vault.is_encrypted(data))
class TestVaultIsEncryptedFile(unittest.TestCase):
def test_utf8_not_encrypted(self):
b_data = "foobar".encode('utf8')
b_data_fo = io.BytesIO(b_data)
self.assertFalse(vault.is_encrypted_file(b_data_fo))
def test_utf8_encrypted(self):
data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
b_data = data.encode('utf8')
b_data_fo = io.BytesIO(b_data)
self.assertTrue(vault.is_encrypted_file(b_data_fo))
def test_bytes_not_encrypted(self):
b_data = b"foobar"
b_data_fo = io.BytesIO(b_data)
self.assertFalse(vault.is_encrypted_file(b_data_fo))
def test_bytes_encrypted(self):
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" + hexlify(b"ansible")
b_data_fo = io.BytesIO(b_data)
self.assertTrue(vault.is_encrypted_file(b_data_fo))
class TestVaultCipherAes256(unittest.TestCase):
def test(self):
vault_cipher = vault.VaultAES256()
self.assertIsInstance(vault_cipher, vault.VaultAES256)
# TODO: tag these as slow tests
def test_create_key(self):
vault_cipher = vault.VaultAES256()
password = 'hunter42'
b_salt = os.urandom(32)
b_key = vault_cipher.create_key(password=password, salt=b_salt, keylength=32, ivlength=16)
self.assertIsInstance(b_key, six.binary_type)
def test_create_key_known(self):
vault_cipher = vault.VaultAES256()
password = 'hunter42'
# A fixed salt
b_salt = b'q' * 32 # q is the most random letter.
b_key = vault_cipher.create_key(password=password, salt=b_salt, keylength=32, ivlength=16)
self.assertIsInstance(b_key, six.binary_type)
# verify we get the same answer
# we could potentially run a few iterations of this and time it to see if it's roughly constant time
# and or that it exceeds some minimal time, but that would likely cause unreliable fails, esp in CI
b_key_2 = vault_cipher.create_key(password=password, salt=b_salt, keylength=32, ivlength=16)
self.assertIsInstance(b_key, six.binary_type)
self.assertEqual(b_key, b_key_2)
def test_is_equal_is_equal(self):
vault_cipher = vault.VaultAES256()
res = vault_cipher.is_equal(b'abcdefghijklmnopqrstuvwxyz', b'abcdefghijklmnopqrstuvwxyz')
self.assertTrue(res)
def test_is_equal_unequal_length(self):
vault_cipher = vault.VaultAES256()
res = vault_cipher.is_equal(b'abcdefghijklmnopqrstuvwxyz', b'abcdefghijklmnopqrstuvwx and sometimes y')
self.assertFalse(res)
def test_is_equal_not_equal(self):
vault_cipher = vault.VaultAES256()
res = vault_cipher.is_equal(b'abcdefghijklmnopqrstuvwxyz', b'AbcdefghijKlmnopQrstuvwxZ')
self.assertFalse(res)
def test_is_equal_empty(self):
vault_cipher = vault.VaultAES256()
res = vault_cipher.is_equal(b'', b'')
self.assertTrue(res)
# NOTE: I'm not really sure what the method should do if it doesn't get bytes,
# but this at least sees if it explodes (maybe it should?)
def test_is_equal_unicode_py3(self):
if not six.PY3:
raise SkipTest
vault_cipher = vault.VaultAES256()
self.assertRaises(TypeError, vault_cipher.is_equal,
u'私はガラスを食べられます。それは私を傷つけません。',
u'私はガラスを食べられます。それは私を傷つけません。')
def test_is_equal_unicode_py2(self):
if not six.PY2:
raise SkipTest
vault_cipher = vault.VaultAES256()
res = vault_cipher.is_equal(u'私はガラスを食べられます。それは私を傷つけません。',
u'私はガラスを食べられます。それは私を傷つけません。')
self.assertTrue(res)
def test_is_equal_unicode_different(self):
vault_cipher = vault.VaultAES256()
res = vault_cipher.is_equal(u'私はガラスを食べられます。それは私を傷つけません。',
u'Pot să mănânc sticlă și ea nu mă rănește.')
self.assertFalse(res)
class TestVaultLib(unittest.TestCase):
def test_methods_exist(self):
@ -69,10 +213,24 @@ class TestVaultLib(unittest.TestCase):
for slot in slots:
assert hasattr(v, slot), "VaultLib is missing the %s method" % slot
def test_encrypt(self):
v = VaultLib(password='the_unit_test_password')
plaintext = u'Some text to encrypt.'
ciphertext = v.encrypt(plaintext)
self.assertIsInstance(ciphertext, (bytes, str))
# TODO: assert something...
def test_is_encrypted(self):
v = VaultLib(None)
assert not v.is_encrypted(u"foobar"), "encryption check on plaintext failed"
assert not v.is_encrypted("foobar".encode('utf-8')), "encryption check on plaintext failed"
data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
assert v.is_encrypted(data.encode('utf-8')), "encryption check on headered text failed"
def test_is_encrypted_bytes(self):
v = VaultLib(None)
assert not v.is_encrypted(b"foobar"), "encryption check on plaintext failed"
data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" + hexlify(b"ansible")
assert v.is_encrypted(data), "encryption check on headered text failed"
def test_format_output(self):
@ -115,35 +273,82 @@ class TestVaultLib(unittest.TestCase):
raise SkipTest
v = VaultLib('ansible')
v.cipher_name = 'AES256'
enc_data = v.encrypt(b"foobar")
plaintext = "foobar"
enc_data = v.encrypt(plaintext)
dec_data = v.decrypt(enc_data)
assert enc_data != b"foobar", "encryption failed"
assert dec_data == b"foobar", "decryption failed"
def test_encrypt_decrypt_aes256_existing_vault(self):
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest
v = VaultLib('test-vault-password')
v.cipher_name = 'AES256'
plaintext = b"Setec Astronomy"
enc_data = '''$ANSIBLE_VAULT;1.1;AES256
33363965326261303234626463623963633531343539616138316433353830356566396130353436
3562643163366231316662386565383735653432386435610a306664636137376132643732393835
63383038383730306639353234326630666539346233376330303938323639306661313032396437
6233623062366136310a633866373936313238333730653739323461656662303864663666653563
3138'''
dec_data = v.decrypt(enc_data)
assert dec_data == plaintext, "decryption failed"
def test_encrypt_decrypt_aes256_bad_hmac(self):
# FIXME This test isn't working quite yet.
raise SkipTest
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest
v = VaultLib('test-vault-password')
v.cipher_name = 'AES256'
# plaintext = "Setec Astronomy"
enc_data = '''$ANSIBLE_VAULT;1.1;AES256
33363965326261303234626463623963633531343539616138316433353830356566396130353436
3562643163366231316662386565383735653432386435610a306664636137376132643732393835
63383038383730306639353234326630666539346233376330303938323639306661313032396437
6233623062366136310a633866373936313238333730653739323461656662303864663666653563
3138'''
b_data = to_bytes(enc_data, errors='strict', encoding='utf-8')
b_data = v._split_header(b_data)
foo = binascii.unhexlify(b_data)
lines = foo.splitlines()
# line 0 is salt, line 1 is hmac, line 2+ is ciphertext
b_salt = lines[0]
b_hmac = lines[1]
b_ciphertext_data = b'\n'.join(lines[2:])
b_ciphertext = binascii.unhexlify(b_ciphertext_data)
# b_orig_ciphertext = b_ciphertext[:]
# now muck with the text
# b_munged_ciphertext = b_ciphertext[:10] + b'\x00' + b_ciphertext[11:]
# b_munged_ciphertext = b_ciphertext
# assert b_orig_ciphertext != b_munged_ciphertext
b_ciphertext_data = binascii.hexlify(b_ciphertext)
b_payload = b'\n'.join([b_salt, b_hmac, b_ciphertext_data])
# reformat
b_invalid_ciphertext = v._format_output(b_payload)
# assert we throw an error
v.decrypt(b_invalid_ciphertext)
def test_encrypt_encrypted(self):
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest
v = VaultLib('ansible')
v.cipher_name = 'AES'
data = "$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(six.b("ansible"))
error_hit = False
try:
enc_data = v.encrypt(data)
except errors.AnsibleError as e:
error_hit = True
assert error_hit, "No error was thrown when trying to encrypt data with a header"
self.assertRaises(errors.AnsibleError, v.encrypt, data,)
def test_decrypt_decrypted(self):
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest
v = VaultLib('ansible')
data = "ansible"
error_hit = False
try:
dec_data = v.decrypt(data)
except errors.AnsibleError as e:
error_hit = True
assert error_hit, "No error was thrown when trying to decrypt data without a header"
self.assertRaises(errors.AnsibleError, v.decrypt, data)
def test_cipher_not_set(self):
# not setting the cipher should default to AES256
@ -151,10 +356,5 @@ class TestVaultLib(unittest.TestCase):
raise SkipTest
v = VaultLib('ansible')
data = "ansible"
error_hit = False
try:
enc_data = v.encrypt(data)
except errors.AnsibleError as e:
error_hit = True
assert not error_hit, "An error was thrown when trying to encrypt data without the cipher set"
assert v.cipher_name == "AES256", "cipher name is not set to AES256: %s" % v.cipher_name
v.encrypt(data)
self.assertEquals(v.cipher_name, "AES256")

View file

@ -22,13 +22,8 @@ __metaclass__ = type
#!/usr/bin/env python
import sys
import getpass
import os
import shutil
import time
import tempfile
from binascii import unhexlify
from binascii import hexlify
from nose.plugins.skip import SkipTest
from ansible.compat.tests import unittest

View file

@ -0,0 +1,64 @@
# coding: utf-8
# 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 io
import yaml
try:
from _yaml import ParserError
except ImportError:
from yaml.parser import ParserError
from ansible.parsing.yaml import dumper
from ansible.parsing.yaml.loader import AnsibleLoader
from ansible.compat.tests import unittest
from ansible.parsing.yaml import objects
from ansible.parsing import vault
from units.mock.yaml_helper import YamlTestUtils
class TestAnsibleDumper(unittest.TestCase, YamlTestUtils):
def setUp(self):
self.vault_password = "hunter42"
self.good_vault = vault.VaultLib(self.vault_password)
self.vault = self.good_vault
self.stream = self._build_stream()
self.dumper = dumper.AnsibleDumper
def _build_stream(self,yaml_text=None):
text = yaml_text or u''
stream = io.StringIO(text)
return stream
def _loader(self, stream):
return AnsibleLoader(stream, vault_password=self.vault_password)
def test(self):
plaintext = 'This is a string we are going to encrypt.'
avu = objects.AnsibleVaultEncryptedUnicode.from_plaintext(plaintext, vault=self.vault)
yaml_out = self._dump_string(avu, dumper=self.dumper)
stream = self._build_stream(yaml_out)
loader = self._loader(stream)
data_from_yaml = loader.get_single_data()
self.assertEquals(plaintext, data_from_yaml.data)

View file

@ -26,20 +26,26 @@ from six import text_type, binary_type
from collections import Sequence, Set, Mapping
from ansible.compat.tests import unittest
from ansible.compat.tests.mock import patch
from ansible import errors
from ansible.parsing.yaml.loader import AnsibleLoader
from ansible.parsing import vault
from ansible.parsing.yaml.objects import AnsibleVaultEncryptedUnicode
from ansible.parsing.yaml.dumper import AnsibleDumper
from ansible.utils.unicode import to_bytes
from units.mock.yaml_helper import YamlTestUtils
try:
from _yaml import ParserError
except ImportError:
from yaml.parser import ParserError
class NameStringIO(StringIO):
"""In py2.6, StringIO doesn't let you set name because a baseclass has it
as readonly property"""
name = None
def __init__(self, *args, **kwargs):
super(NameStringIO, self).__init__(*args, **kwargs)
@ -159,6 +165,124 @@ class TestAnsibleLoaderBasic(unittest.TestCase):
self.assertEqual(data[0][u'baz'].ansible_pos, ('myfile.yml', 2, 9))
class TestAnsibleLoaderVault(unittest.TestCase, YamlTestUtils):
def setUp(self):
self.vault_password = "hunter42"
self.vault = vault.VaultLib(self.vault_password)
def test_wrong_password(self):
plaintext = u"Ansible"
bob_password = "this is a different password"
bobs_vault = vault.VaultLib(bob_password)
ciphertext = bobs_vault.encrypt(plaintext)
try:
self.vault.decrypt(ciphertext)
except Exception as e:
self.assertIsInstance(e, errors.AnsibleError)
self.assertEqual(e.message, 'Decryption failed')
def _encrypt_plaintext(self, plaintext):
# Construct a yaml repr of a vault by hand
vaulted_var_bytes = self.vault.encrypt(plaintext)
# add yaml tag
vaulted_var = vaulted_var_bytes.decode()
lines = vaulted_var.splitlines()
lines2 = []
for line in lines:
lines2.append(' %s' % line)
vaulted_var = '\n'.join(lines2)
tagged_vaulted_var = u"""!vault-encrypted |\n%s""" % vaulted_var
return tagged_vaulted_var
def _build_stream(self, yaml_text):
stream = NameStringIO(yaml_text)
stream.name = 'my.yml'
return stream
def _loader(self, stream):
return AnsibleLoader(stream, vault_password=self.vault_password)
def _load_yaml(self, yaml_text, password):
stream = self._build_stream(yaml_text)
loader = self._loader(stream)
data_from_yaml = loader.get_single_data()
return data_from_yaml
def test_dump_load_cycle(self):
avu = AnsibleVaultEncryptedUnicode.from_plaintext('The plaintext for test_dump_load_cycle.', vault=self.vault)
self._dump_load_cycle(avu)
def test_embedded_vault_from_dump(self):
avu = AnsibleVaultEncryptedUnicode.from_plaintext('setec astronomy', vault=self.vault)
blip = {'stuff1': [{'a dict key': 24},
{'shhh-ssh-secrets': avu,
'nothing to see here': 'move along'}],
'another key': 24.1}
blip = ['some string', 'another string', avu]
stream = NameStringIO()
self._dump_stream(blip, stream, dumper=AnsibleDumper)
print(stream.getvalue())
stream.seek(0)
stream.seek(0)
loader = self._loader(stream)
data_from_yaml = loader.get_data()
stream2 = NameStringIO(u'')
# verify we can dump the object again
self._dump_stream(data_from_yaml, stream2, dumper=AnsibleDumper)
def test_embedded_vault(self):
plaintext_var = u"""This is the plaintext string."""
tagged_vaulted_var = self._encrypt_plaintext(plaintext_var)
another_vaulted_var = self._encrypt_plaintext(plaintext_var)
different_var = u"""A different string that is not the same as the first one."""
different_vaulted_var = self._encrypt_plaintext(different_var)
yaml_text = u"""---\nwebster: daniel\noed: oxford\nthe_secret: %s\nanother_secret: %s\ndifferent_secret: %s""" % (tagged_vaulted_var, another_vaulted_var, different_vaulted_var)
data_from_yaml = self._load_yaml(yaml_text, self.vault_password)
vault_string = data_from_yaml['the_secret']
self.assertEquals(plaintext_var, data_from_yaml['the_secret'])
test_dict = {}
test_dict[vault_string] = 'did this work?'
self.assertEquals(vault_string.data, vault_string)
# This looks weird and useless, but the object in question has a custom __eq__
self.assertEquals(vault_string, vault_string)
another_vault_string = data_from_yaml['another_secret']
different_vault_string = data_from_yaml['different_secret']
self.assertEquals(vault_string, another_vault_string)
self.assertNotEquals(vault_string, different_vault_string)
# More testing of __eq__/__ne__
self.assertTrue('some string' != vault_string)
self.assertNotEquals('some string', vault_string)
# Note this is a compare of the str/unicode of these, they are diferent types
# so we want to test self == other, and other == self etc
self.assertEquals(plaintext_var, vault_string)
self.assertEquals(vault_string, plaintext_var)
self.assertFalse(plaintext_var != vault_string)
self.assertFalse(vault_string != plaintext_var)
class TestAnsibleLoaderPlay(unittest.TestCase):
def setUp(self):
@ -242,7 +366,7 @@ class TestAnsibleLoaderPlay(unittest.TestCase):
def check_vars(self):
# Numbers don't have line/col information yet
#self.assertEqual(self.data[0][u'vars'][u'number'].ansible_pos, (self.play_filename, 4, 21))
# self.assertEqual(self.data[0][u'vars'][u'number'].ansible_pos, (self.play_filename, 4, 21))
self.assertEqual(self.data[0][u'vars'][u'string'].ansible_pos, (self.play_filename, 5, 29))
self.assertEqual(self.data[0][u'vars'][u'utf8_string'].ansible_pos, (self.play_filename, 6, 34))
@ -255,8 +379,8 @@ class TestAnsibleLoaderPlay(unittest.TestCase):
self.assertEqual(self.data[0][u'vars'][u'list'][0].ansible_pos, (self.play_filename, 11, 25))
self.assertEqual(self.data[0][u'vars'][u'list'][1].ansible_pos, (self.play_filename, 12, 25))
# Numbers don't have line/col info yet
#self.assertEqual(self.data[0][u'vars'][u'list'][2].ansible_pos, (self.play_filename, 13, 25))
#self.assertEqual(self.data[0][u'vars'][u'list'][3].ansible_pos, (self.play_filename, 14, 25))
# self.assertEqual(self.data[0][u'vars'][u'list'][2].ansible_pos, (self.play_filename, 13, 25))
# self.assertEqual(self.data[0][u'vars'][u'list'][3].ansible_pos, (self.play_filename, 14, 25))
def check_tasks(self):
#

View file

@ -0,0 +1,129 @@
# 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/>.
#
# Copyright 2016, Adrian Likins <alikins@redhat.com>
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible.compat.tests import unittest
from ansible.parsing import vault
from ansible.parsing.yaml.loader import AnsibleLoader
# module under test
from ansible.parsing.yaml import objects
from units.mock.yaml_helper import YamlTestUtils
class TestAnsibleVaultUnicodeNoVault(unittest.TestCase, YamlTestUtils):
def test_empty_init(self):
self.assertRaises(TypeError, objects.AnsibleVaultEncryptedUnicode)
def test_empty_string_init(self):
seq = ''.encode('utf8')
self.assert_values(seq)
def test_empty_byte_string_init(self):
seq = b''
self.assert_values(seq)
def _assert_values(self, avu, seq):
self.assertIsInstance(avu, objects.AnsibleVaultEncryptedUnicode)
self.assertTrue(avu.vault is None)
# AnsibleVaultEncryptedUnicode without a vault should never == any string
self.assertNotEquals(avu, seq)
def assert_values(self, seq):
avu = objects.AnsibleVaultEncryptedUnicode(seq)
self._assert_values(avu, seq)
def test_single_char(self):
seq = 'a'.encode('utf8')
self.assert_values(seq)
def test_string(self):
seq = 'some letters'
self.assert_values(seq)
def test_byte_string(self):
seq = 'some letters'.encode('utf8')
self.assert_values(seq)
class TestAnsibleVaultEncryptedUnicode(unittest.TestCase, YamlTestUtils):
def setUp(self):
self.vault_password = "hunter42"
self.good_vault = vault.VaultLib(self.vault_password)
self.wrong_vault_password = 'not-hunter42'
self.wrong_vault = vault.VaultLib(self.wrong_vault_password)
self.vault = self.good_vault
def _loader(self, stream):
return AnsibleLoader(stream, vault_password=self.vault_password)
def test_dump_load_cycle(self):
aveu = self._from_plaintext('the test string for TestAnsibleVaultEncryptedUnicode.test_dump_load_cycle')
self._dump_load_cycle(aveu)
def assert_values(self, avu, seq):
self.assertIsInstance(avu, objects.AnsibleVaultEncryptedUnicode)
self.assertEquals(avu, seq)
self.assertTrue(avu.vault is self.vault)
self.assertIsInstance(avu.vault, vault.VaultLib)
def _from_plaintext(self, seq):
return objects.AnsibleVaultEncryptedUnicode.from_plaintext(seq, vault=self.vault)
def _from_ciphertext(self, ciphertext):
avu = objects.AnsibleVaultEncryptedUnicode(ciphertext)
avu.vault = self.vault
return avu
def test_empty_init(self):
self.assertRaises(TypeError, objects.AnsibleVaultEncryptedUnicode)
def test_empty_string_init_from_plaintext(self):
seq = ''
avu = self._from_plaintext(seq)
self.assert_values(avu,seq)
def test_empty_unicode_init_from_plaintext(self):
seq = u''
avu = self._from_plaintext(seq)
self.assert_values(avu,seq)
def test_string_from_plaintext(self):
seq = 'some letters'
avu = self._from_plaintext(seq)
self.assert_values(avu,seq)
def test_unicode_from_plaintext(self):
seq = u'some letters'
avu = self._from_plaintext(seq)
self.assert_values(avu,seq)
# TODO/FIXME: make sure bad password fails differently than 'thats not encrypted'
def test_empty_string_wrong_password(self):
seq = ''
self.vault = self.wrong_vault
avu = self._from_plaintext(seq)
self.assert_values(avu, seq)