2014-10-21 18:27:01 +00:00
|
|
|
# (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 getpass
|
|
|
|
import os
|
|
|
|
import shutil
|
|
|
|
import time
|
|
|
|
import tempfile
|
2015-04-15 04:34:30 +00:00
|
|
|
import six
|
|
|
|
|
2014-10-21 18:27:01 +00:00
|
|
|
from binascii import unhexlify
|
|
|
|
from binascii import hexlify
|
|
|
|
from nose.plugins.skip import SkipTest
|
|
|
|
|
|
|
|
from ansible.compat.tests import unittest
|
2015-04-15 18:08:53 +00:00
|
|
|
from ansible.utils.unicode import to_bytes, to_unicode
|
2014-10-21 18:27:01 +00:00
|
|
|
|
|
|
|
from ansible import errors
|
|
|
|
from ansible.parsing.vault import VaultLib
|
|
|
|
|
|
|
|
# Counter import fails for 2.0.1, requires >= 2.6.1 from pip
|
|
|
|
try:
|
|
|
|
from Crypto.Util import Counter
|
|
|
|
HAS_COUNTER = True
|
|
|
|
except ImportError:
|
|
|
|
HAS_COUNTER = False
|
|
|
|
|
|
|
|
# KDF import fails for 2.0.1, requires >= 2.6.1 from pip
|
|
|
|
try:
|
|
|
|
from Crypto.Protocol.KDF import PBKDF2
|
|
|
|
HAS_PBKDF2 = True
|
|
|
|
except ImportError:
|
|
|
|
HAS_PBKDF2 = False
|
|
|
|
|
|
|
|
# AES IMPORTS
|
|
|
|
try:
|
|
|
|
from Crypto.Cipher import AES as AES
|
|
|
|
HAS_AES = True
|
|
|
|
except ImportError:
|
|
|
|
HAS_AES = False
|
|
|
|
|
|
|
|
class TestVaultLib(unittest.TestCase):
|
|
|
|
|
|
|
|
def test_methods_exist(self):
|
|
|
|
v = VaultLib('ansible')
|
|
|
|
slots = ['is_encrypted',
|
|
|
|
'encrypt',
|
|
|
|
'decrypt',
|
2015-08-24 22:49:55 +00:00
|
|
|
'_format_output',
|
2014-10-21 18:27:01 +00:00
|
|
|
'_split_header',]
|
2015-04-15 04:34:30 +00:00
|
|
|
for slot in slots:
|
2014-10-21 18:27:01 +00:00
|
|
|
assert hasattr(v, slot), "VaultLib is missing the %s method" % slot
|
|
|
|
|
|
|
|
def test_is_encrypted(self):
|
|
|
|
v = VaultLib(None)
|
2015-04-15 18:08:53 +00:00
|
|
|
assert not v.is_encrypted(u"foobar"), "encryption check on plaintext failed"
|
|
|
|
data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
|
2014-10-21 18:27:01 +00:00
|
|
|
assert v.is_encrypted(data), "encryption check on headered text failed"
|
|
|
|
|
2015-08-24 22:49:55 +00:00
|
|
|
def test_format_output(self):
|
2014-10-21 18:27:01 +00:00
|
|
|
v = VaultLib('ansible')
|
|
|
|
v.cipher_name = "TEST"
|
2015-10-16 06:10:25 +00:00
|
|
|
sensitive_data = b"ansible"
|
2015-08-24 22:49:55 +00:00
|
|
|
data = v._format_output(sensitive_data)
|
2015-04-15 18:08:53 +00:00
|
|
|
lines = data.split(b'\n')
|
2014-10-21 18:27:01 +00:00
|
|
|
assert len(lines) > 1, "failed to properly add header"
|
2015-10-16 06:10:25 +00:00
|
|
|
header = to_bytes(lines[0])
|
|
|
|
assert header.endswith(b';TEST'), "header does end with cipher name"
|
|
|
|
header_parts = header.split(b';')
|
2015-04-15 04:34:30 +00:00
|
|
|
assert len(header_parts) == 3, "header has the wrong number of parts"
|
2015-10-16 06:10:25 +00:00
|
|
|
assert header_parts[0] == b'$ANSIBLE_VAULT', "header does not start with $ANSIBLE_VAULT"
|
2015-08-24 22:49:55 +00:00
|
|
|
assert header_parts[1] == v.b_version, "header version is incorrect"
|
2015-10-16 06:10:25 +00:00
|
|
|
assert header_parts[2] == b'TEST', "header does end with cipher name"
|
2014-10-21 18:27:01 +00:00
|
|
|
|
|
|
|
def test_split_header(self):
|
|
|
|
v = VaultLib('ansible')
|
2015-04-15 18:08:53 +00:00
|
|
|
data = b"$ANSIBLE_VAULT;9.9;TEST\nansible"
|
2015-04-15 04:34:30 +00:00
|
|
|
rdata = v._split_header(data)
|
2015-04-15 18:08:53 +00:00
|
|
|
lines = rdata.split(b'\n')
|
|
|
|
assert lines[0] == b"ansible"
|
2014-10-21 18:27:01 +00:00
|
|
|
assert v.cipher_name == 'TEST', "cipher name was not set"
|
2015-10-16 06:11:34 +00:00
|
|
|
assert v.b_version == b"9.9"
|
2014-10-21 18:27:01 +00:00
|
|
|
|
|
|
|
def test_encrypt_decrypt_aes(self):
|
|
|
|
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
|
|
|
raise SkipTest
|
|
|
|
v = VaultLib('ansible')
|
2015-04-15 18:08:53 +00:00
|
|
|
v.cipher_name = u'AES'
|
2015-08-27 13:06:05 +00:00
|
|
|
# AES encryption code has been removed, so this is old output for
|
|
|
|
# AES-encrypted 'foobar' with password 'ansible'.
|
2015-10-16 06:12:49 +00:00
|
|
|
enc_data = b'$ANSIBLE_VAULT;1.1;AES\n53616c7465645f5fc107ce1ef4d7b455e038a13b053225776458052f8f8f332d554809d3f150bfa3\nfe3db930508b65e0ff5947e4386b79af8ab094017629590ef6ba486814cf70f8e4ab0ed0c7d2587e\n786a5a15efeb787e1958cbdd480d076c\n'
|
2014-10-21 18:27:01 +00:00
|
|
|
dec_data = v.decrypt(enc_data)
|
2015-10-16 06:12:49 +00:00
|
|
|
assert dec_data == b"foobar", "decryption failed"
|
2014-10-21 18:27:01 +00:00
|
|
|
|
|
|
|
def test_encrypt_decrypt_aes256(self):
|
|
|
|
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
|
|
|
raise SkipTest
|
|
|
|
v = VaultLib('ansible')
|
|
|
|
v.cipher_name = 'AES256'
|
2015-10-16 06:13:46 +00:00
|
|
|
enc_data = v.encrypt(b"foobar")
|
2014-10-21 18:27:01 +00:00
|
|
|
dec_data = v.decrypt(enc_data)
|
2015-10-16 06:13:46 +00:00
|
|
|
assert enc_data != b"foobar", "encryption failed"
|
|
|
|
assert dec_data == b"foobar", "decryption failed"
|
2014-10-21 18:27:01 +00:00
|
|
|
|
|
|
|
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'
|
2015-04-15 04:34:30 +00:00
|
|
|
data = "$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(six.b("ansible"))
|
2014-10-21 18:27:01 +00:00
|
|
|
error_hit = False
|
|
|
|
try:
|
|
|
|
enc_data = v.encrypt(data)
|
2014-10-29 00:27:14 +00:00
|
|
|
except errors.AnsibleError as e:
|
2014-10-21 18:27:01 +00:00
|
|
|
error_hit = True
|
2015-04-15 04:34:30 +00:00
|
|
|
assert error_hit, "No error was thrown when trying to encrypt data with a header"
|
2014-10-21 18:27:01 +00:00
|
|
|
|
|
|
|
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)
|
2014-10-29 00:27:14 +00:00
|
|
|
except errors.AnsibleError as e:
|
2014-10-21 18:27:01 +00:00
|
|
|
error_hit = True
|
2015-04-15 04:34:30 +00:00
|
|
|
assert error_hit, "No error was thrown when trying to decrypt data without a header"
|
2014-10-21 18:27:01 +00:00
|
|
|
|
|
|
|
def test_cipher_not_set(self):
|
|
|
|
# not setting the cipher should default to AES256
|
|
|
|
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
|
|
|
raise SkipTest
|
|
|
|
v = VaultLib('ansible')
|
|
|
|
data = "ansible"
|
|
|
|
error_hit = False
|
|
|
|
try:
|
|
|
|
enc_data = v.encrypt(data)
|
2014-10-29 00:27:14 +00:00
|
|
|
except errors.AnsibleError as e:
|
2014-10-21 18:27:01 +00:00
|
|
|
error_hit = True
|
2015-04-15 04:34:30 +00:00
|
|
|
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
|