archive
/
ansible
mirror of https://github.com/ansible/ansible.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #17590 from abadger/vault-fixes

Browse Source 
Vault fixes
pull/17607/head
Toshio Kuratomi 8 years ago committed by GitHub
parent 355b29573c e70066a6f7
commit 8af8eec789
8 changed files with 385 additions and 370 deletions
Show Stats Download Patch File Download Diff File

16
CHANGELOG.md
Unescape Escape View File

@ -169,6 +169,22 @@ Ansible Changes By Release
* removed previously deprecated ';' as host list separator. * removed previously deprecated ';' as host list separator.
* Only check if the default ssh client supports ControlPersist once instead of once for each host + task combination. * Only check if the default ssh client supports ControlPersist once instead of once for each host + task combination.
###For custom front ends using the API:
* ansible.parsing.vault:
* VaultLib.is_encrypted() has been deprecated. It will be removed in 2.4.
Use ansible.parsing.vault.is_encrypted() instead
* VaultFile has been removed. This unfinished code was never used inside of
Ansible. The feature it was intended to support has now been implemented
without using this.
* VaultAES, the older, insecure encrypted format that debuted in Ansible-1.5
and was relaced by VaultAES256 less than a week later, now has a deprecation
warning. **It will be removed in 2.3**. In the unlikely event that you
wrote a vault file in that 1 week window and have never modified the file
since (ansible-vault automatically re-encrypts the file using VaultAES256
whenever it is written to but not read), run ``ansible-vault rekey
[filename]`` to move to VaultAES256.
## 2.1.2 "The Song Remains the Same" ## 2.1.2 "The Song Remains the Same"
###Deprecations: ###Deprecations:

366
lib/ansible/parsing/vault/__init__.py
Unescape Escape View File

@ -1,4 +1,6 @@
# (c) 2014, James Tanner <tanner.jc@gmail.com> # (c) 2014, James Tanner <tanner.jc@gmail.com>
# (c) 2016, Adrian Likins <alikins@redhat.com>
# (c) 2016, Toshio Kuratomi <tkuratomi@ansible.com>
# #
# Ansible is free software: you can redistribute it and/or modify # Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by # it under the terms of the GNU General Public License as published by
@ -25,7 +27,6 @@ import tempfile
import random import random
from io import BytesIO from io import BytesIO
from subprocess import call from subprocess import call
from ansible.errors import AnsibleError
from hashlib import sha256 from hashlib import sha256
from binascii import hexlify from binascii import hexlify
from binascii import unhexlify from binascii import unhexlify
@ -61,7 +62,9 @@ try:
except ImportError: except ImportError:
HAS_AES = False HAS_AES = False
from ansible.compat.six import PY3 from ansible.compat.six import PY3, binary_type
from ansible.compat.six.moves import zip
from ansible.errors import AnsibleError
from ansible.module_utils._text import to_bytes, to_text from ansible.module_utils._text import to_bytes, to_text
try: try:
@ -84,14 +87,13 @@ except Exception as e:
import traceback import traceback
display.debug("Traceback from import of cryptography was {0}".format(traceback.format_exc())) display.debug("Traceback from import of cryptography was {0}".format(traceback.format_exc()))
HAS_ANY_PBKDF2HMAC = HAS_PBKDF2 or HAS_PBKDF2HMAC HAS_ANY_PBKDF2HMAC = HAS_PBKDF2 or HAS_PBKDF2HMAC
CRYPTO_UPGRADE = "ansible-vault requires a newer version of pycrypto than the one installed on your platform." \ CRYPTO_UPGRADE = "ansible-vault requires a newer version of pycrypto than the one installed on your platform." \
" You may fix this with OS-specific commands such as: yum install python-devel; rpm -e --nodeps python-crypto; pip install pycrypto" " You may fix this with OS-specific commands such as: yum install python-devel; rpm -e --nodeps python-crypto; pip install pycrypto"
b_HEADER = b'$ANSIBLE_VAULT' b_HEADER = b'$ANSIBLE_VAULT'
HEADER = '$ANSIBLE_VAULT'
CIPHER_WHITELIST = frozenset((u'AES', u'AES256')) CIPHER_WHITELIST = frozenset((u'AES', u'AES256'))
CIPHER_WRITE_WHITELIST = frozenset((u'AES256',)) CIPHER_WRITE_WHITELIST = frozenset((u'AES256',))
# See also CIPHER_MAPPING at the bottom of the file which maps cipher strings # See also CIPHER_MAPPING at the bottom of the file which maps cipher strings
@ -108,32 +110,56 @@ class AnsibleVaultError(AnsibleError):
pass pass
def is_encrypted(b_data): def is_encrypted(data):
""" Test if this is vault encrypted data blob """ Test if this is vault encrypted data blob
:arg data: a python2 str or a python3 'bytes' to test whether it is :arg data: a byte or text string to test whether it is recognized as vault
recognized as vault encrypted data encrypted data
:returns: True if it is recognized. Otherwise, False. :returns: True if it is recognized. Otherwise, False.
""" """
try:
# Make sure we have a byte string and that it only contains ascii
# bytes.
b_data = to_bytes(to_text(data, encoding='ascii', errors='strict', nonstring='strict'), encoding='ascii', errors='strict')
except (UnicodeError, TypeError):
# The vault format is pure ascii so if we failed to encode to bytes
# via ascii we know that this is not vault data.
# Similarly, if it's not a string, it's not vault data
return False
if b_data.startswith(b_HEADER): if b_data.startswith(b_HEADER):
return True return True
return False return False
def is_encrypted_file(file_obj): def is_encrypted_file(file_obj, start_pos=0, count=-1):
"""Test if the contents of a file obj are a vault encrypted data blob. """Test if the contents of a file obj are a vault encrypted data blob.
The data read from the file_obj is expected to be bytestrings (py2 'str' or
python3 'bytes'). This more or less expects 'utf-8' encoding.
:arg file_obj: A file object that will be read from. :arg file_obj: A file object that will be read from.
:returns: True if the file is a vault file. Otherwise, False. :kwarg start_pos: A byte offset in the file to start reading the header
from. Defaults to 0, the beginning of the file.
:kwarg count: Read up to this number of bytes from the file to determine
if it looks like encrypted vault data. The default is -1, read to the
end of file.
:returns: True if the file looks like a vault file. Otherwise, False.
""" """
# read the header and reset the file stream to where it started # read the header and reset the file stream to where it started
current_position = file_obj.tell() current_position = file_obj.tell()
b_header_part = file_obj.read(len(b_HEADER)) try:
file_obj.seek(current_position) file_obj.seek(start_pos)
return is_encrypted(b_header_part) vaulttext = file_obj.read(count)
try:
b_vaulttext = to_bytes(to_text(vaulttext, encoding='ascii', errors='strict'), encoding='ascii', errors='strict')
except (UnicodeError, TypeError):
# At present, vault files contain only ascii characters. The encoding is utf-8
# without BOM (for future expansion). If the header does not
# decode as ascii then we know we do not have proper vault
# encrypted data.
return False
finally:
file_obj.seek(current_position)
return is_encrypted(b_vaulttext)
class VaultLib: class VaultLib:
@ -143,11 +169,11 @@ class VaultLib:
self.cipher_name = None self.cipher_name = None
self.b_version = b'1.1' self.b_version = b'1.1'
# really b_data, but for compat @staticmethod
def is_encrypted(self, data): def is_encrypted(data):
""" Test if this is vault encrypted data """ Test if this is vault encrypted data
:arg data: a python2 utf-8 string or a python3 'bytes' to test whether it is :arg data: a byte or text string or a python3 to test for whether it is
recognized as vault encrypted data recognized as vault encrypted data
:returns: True if it is recognized. Otherwise, False. :returns: True if it is recognized. Otherwise, False.
""" """
@ -155,36 +181,29 @@ class VaultLib:
# This could in the future, check to see if the data is a vault blob and # This could in the future, check to see if the data is a vault blob and
# is encrypted with a key associated with this vault # is encrypted with a key associated with this vault
# instead of just checking the format. # instead of just checking the format.
display.deprecated(u'vault.VaultLib.is_encrypted is deprecated. Use vault.is_encrypted instead', version='2.4')
return is_encrypted(data) return is_encrypted(data)
def is_encrypted_file(self, file_obj): @staticmethod
def is_encrypted_file(file_obj):
display.deprecated(u'vault.VaultLib.is_encrypted_file is deprecated. Use vault.is_encrypted_file instead', version='2.4')
return is_encrypted_file(file_obj) return is_encrypted_file(file_obj)
def encrypt(self, data): def encrypt(self, plaintext):
"""Vault encrypt a piece of data. """Vault encrypt a piece of data.
:arg data: a PY2 unicode string or PY3 string to encrypt. :arg plaintext: a text or byte string to encrypt.
:returns: a utf-8 encoded byte str of encrypted data. The string :returns: a utf-8 encoded byte str of encrypted data. The string
contains a header identifying this as vault encrypted data and contains a header identifying this as vault encrypted data and
formatted to newline terminated lines of 80 characters. This is formatted to newline terminated lines of 80 characters. This is
suitable for dumping as is to a vault file. suitable for dumping as is to a vault file.
The unicode or string passed in as data will encoded to UTF-8 before If the string passed in is a text string, it will be encoded to UTF-8
encryption. If the a already encoded string or PY2 bytestring needs to before encryption.
be encrypted, use encrypt_bytestring().
""" """
plaintext = data b_plaintext = to_bytes(plaintext, errors='surrogate_or_strict')
plaintext_bytes = plaintext.encode('utf-8')
return self.encrypt_bytestring(plaintext_bytes)
def encrypt_bytestring(self, plaintext_bytes):
'''Encrypt a PY2 bytestring.
Like encrypt(), except plaintext_bytes is not encoded to UTF-8
before encryption.'''
if self.is_encrypted(plaintext_bytes): if is_encrypted(b_plaintext):
raise AnsibleError("input is already encrypted") raise AnsibleError("input is already encrypted")
if not self.cipher_name or self.cipher_name not in CIPHER_WRITE_WHITELIST: if not self.cipher_name or self.cipher_name not in CIPHER_WRITE_WHITELIST:
@ -197,32 +216,35 @@ class VaultLib:
this_cipher = Cipher() this_cipher = Cipher()
# encrypt data # encrypt data
ciphertext_bytes = this_cipher.encrypt(plaintext_bytes, self.b_password) b_ciphertext = this_cipher.encrypt(b_plaintext, self.b_password)
# format the data for output to the file # format the data for output to the file
ciphertext_envelope = self._format_output(ciphertext_bytes) b_vaulttext = self._format_output(b_ciphertext)
return ciphertext_envelope return b_vaulttext
def decrypt(self, data, filename=None): def decrypt(self, vaulttext, filename=None):
"""Decrypt a piece of vault encrypted data. """Decrypt a piece of vault encrypted data.
:arg data: a string to decrypt. Since vault encrypted data is an :arg vaulttext: a string to decrypt. Since vault encrypted data is an
ascii text format this can be either a byte str or unicode string. ascii text format this can be either a byte str or unicode string.
:kwarg filename: a filename that the data came from. This is only
used to make better error messages in case the data cannot be
decrypted.
:returns: a byte string containing the decrypted data :returns: a byte string containing the decrypted data
""" """
b_data = to_bytes(data, errors='strict', encoding='utf-8') b_vaulttext = to_bytes(vaulttext, errors='strict', encoding='utf-8')
if self.b_password is None: if self.b_password is None:
raise AnsibleError("A vault password must be specified to decrypt data") raise AnsibleError("A vault password must be specified to decrypt data")
if not self.is_encrypted(b_data): if not is_encrypted(b_vaulttext):
msg = "input is not vault encrypted data" msg = "input is not vault encrypted data"
if filename: if filename:
msg += "%s is not a vault encrypted file" % filename msg += "%s is not a vault encrypted file" % filename
raise AnsibleError(msg) raise AnsibleError(msg)
# clean out header # clean out header
b_data = self._split_header(b_data) b_vaulttext = self._split_header(b_vaulttext)
# create the cipher object # create the cipher object
cipher_class_name = u'Vault{0}'.format(self.cipher_name) cipher_class_name = u'Vault{0}'.format(self.cipher_name)
@ -233,20 +255,20 @@ class VaultLib:
else: else:
raise AnsibleError("{0} cipher could not be found".format(self.cipher_name)) raise AnsibleError("{0} cipher could not be found".format(self.cipher_name))
# try to unencrypt data # try to unencrypt vaulttext
b_data = this_cipher.decrypt(b_data, self.b_password) b_plaintext = this_cipher.decrypt(b_vaulttext, self.b_password)
if b_data is None: if b_plaintext is None:
msg = "Decryption failed" msg = "Decryption failed"
if filename: if filename:
msg += " on %s" % filename msg += " on %s" % filename
raise AnsibleError(msg) raise AnsibleError(msg)
return b_data return b_plaintext
def _format_output(self, b_data): def _format_output(self, b_ciphertext):
""" Add header and format to 80 columns """ Add header and format to 80 columns
:arg b_data: the encrypted and hexlified data as a byte string :arg b_vaulttext: the encrypted and hexlified data as a byte string
:returns: a byte str that should be dumped into a file. It's :returns: a byte str that should be dumped into a file. It's
formatted to 80 char columns and has the header prepended formatted to 80 char columns and has the header prepended
""" """
@ -254,38 +276,37 @@ class VaultLib:
if not self.cipher_name: if not self.cipher_name:
raise AnsibleError("the cipher must be set before adding a header") raise AnsibleError("the cipher must be set before adding a header")
b_header = HEADER.encode('utf-8') header = b';'.join([b_HEADER, self.b_version,
header = b';'.join([b_header, self.b_version,
to_bytes(self.cipher_name,'utf-8', errors='strict')]) to_bytes(self.cipher_name,'utf-8', errors='strict')])
tmpdata = [header] b_vaulttext = [header]
tmpdata += [b_data[i:i + 80] for i in range(0, len(b_data), 80)] b_vaulttext += [b_ciphertext[i:i + 80] for i in range(0, len(b_ciphertext), 80)]
tmpdata += [b''] b_vaulttext += [b'']
tmpdata = b'\n'.join(tmpdata) b_vaulttext = b'\n'.join(b_vaulttext)
return tmpdata return b_vaulttext
def _split_header(self, b_data): def _split_header(self, b_vaulttext):
"""Retrieve information about the Vault and clean the data """Retrieve information about the Vault and clean the data
When data is saved, it has a header prepended and is formatted into 80 When data is saved, it has a header prepended and is formatted into 80
character lines. This method extracts the information from the header character lines. This method extracts the information from the header
and then removes the header and the inserted newlines. The string returned and then removes the header and the inserted newlines. The string returned
is suitable for processing by the Cipher classes. is suitable for processing by the Cipher classes.
:arg b_data: byte str containing the data from a save file :arg b_vaulttext: byte str containing the data from a save file
:returns: a byte str suitable for passing to a Cipher class's :returns: a byte str suitable for passing to a Cipher class's
decrypt() function. decrypt() function.
""" """
# used by decrypt # used by decrypt
tmpdata = b_data.split(b'\n') b_tmpdata = b_vaulttext.split(b'\n')
tmpheader = tmpdata[0].strip().split(b';') b_tmpheader = b_tmpdata[0].strip().split(b';')
self.b_version = tmpheader[1].strip() self.b_version = b_tmpheader[1].strip()
self.cipher_name = to_text(tmpheader[2].strip()) self.cipher_name = to_text(b_tmpheader[2].strip())
clean_data = b''.join(tmpdata[1:]) b_ciphertext = b''.join(b_tmpdata[1:])
return clean_data return b_ciphertext
class VaultEditor: class VaultEditor:
@ -400,7 +421,7 @@ class VaultEditor:
# A file to be encrypted into a vaultfile could be any encoding # A file to be encrypted into a vaultfile could be any encoding
# so treat the contents as a byte string. # so treat the contents as a byte string.
plaintext = self.read_data(filename) plaintext = self.read_data(filename)
ciphertext = self.vault.encrypt_bytestring(plaintext) ciphertext = self.vault.encrypt(plaintext)
self.write_data(ciphertext, output_file or filename) self.write_data(ciphertext, output_file or filename)
def decrypt_file(self, filename, output_file=None): def decrypt_file(self, filename, output_file=None):
@ -533,47 +554,6 @@ class VaultEditor:
return editor return editor
class VaultFile(object):
def __init__(self, password, filename):
self.password = password
self.filename = filename
if not os.path.isfile(self.filename):
raise AnsibleError("%s does not exist" % self.filename)
try:
self.filehandle = open(filename, "rb")
except Exception as e:
raise AnsibleError("Could not open %s: %s" % (self.filename, str(e)))
_, self.tmpfile = tempfile.mkstemp()
# TODO:
# __del__ can be problematic in python... For this use case, make
# VaultFile a context manager instead (implement __enter__ and __exit__)
def __del__(self):
self.filehandle.close()
os.unlink(self.tmpfile)
def is_encrypted(self):
return is_encrypted_file(self.filehandle)
def get_decrypted(self):
check_prereqs()
if self.is_encrypted():
tmpdata = self.filehandle.read()
this_vault = VaultLib(self.password)
dec_data = this_vault.decrypt(tmpdata)
if dec_data is None:
raise AnsibleError("Failed to decrypt: %s" % self.filename)
else:
self.tmpfile.write(dec_data)
return self.tmpfile
else:
return self.filename
######################################## ########################################
# CIPHERS # # CIPHERS #
######################################## ########################################
@ -591,76 +571,84 @@ class VaultAES:
if not HAS_AES: if not HAS_AES:
raise AnsibleError(CRYPTO_UPGRADE) raise AnsibleError(CRYPTO_UPGRADE)
def aes_derive_key_and_iv(self, password, salt, key_length, iv_length): def _aes_derive_key_and_iv(self, b_password, b_salt, key_length, iv_length):
""" Create a key and an initialization vector """ """ Create a key and an initialization vector """
d = d_i = b'' b_d = b_di = b''
while len(d) < key_length + iv_length: while len(b_d) < key_length + iv_length:
text = b''.join([d_i, password, salt]) b_text = b''.join([b_di, b_password, b_salt])
d_i = to_bytes(md5(text).digest(), errors='strict') b_di = to_bytes(md5(b_text).digest(), errors='strict')
d += d_i b_d += b_di
key = d[:key_length] b_key = b_d[:key_length]
iv = d[key_length:key_length+iv_length] b_iv = b_d[key_length:key_length+iv_length]
return key, iv return b_key, b_iv
def encrypt(self, data, password, key_length=32): def encrypt(self, b_plaintext, b_password, key_length=32):
""" Read plaintext data from in_file and write encrypted to out_file """ """ Read plaintext data from in_file and write encrypted to out_file """
raise AnsibleError("Encryption disabled for deprecated VaultAES class") raise AnsibleError("Encryption disabled for deprecated VaultAES class")
def decrypt(self, data, password, key_length=32): def decrypt(self, b_vaulttext, b_password, key_length=32):
""" Read encrypted data from in_file and write decrypted to out_file """ """ Decrypt the given data and return it
:arg b_data: A byte string containing the encrypted data
:arg b_password: A byte string containing the encryption password
:arg key_length: Length of the key
:returns: A byte string containing the decrypted data
"""
display.deprecated(u'The VaultAES format is insecure and has been'
' deprecated since Ansible-1.5. Use vault rekey FILENAME to'
' switch to the newer VaultAES256 format', version='2.3')
# http://stackoverflow.com/a/14989032 # http://stackoverflow.com/a/14989032
data = unhexlify(data) b_ciphertext = unhexlify(b_vaulttext)
in_file = BytesIO(data) in_file = BytesIO(b_ciphertext)
in_file.seek(0) in_file.seek(0)
out_file = BytesIO() out_file = BytesIO()
bs = AES.block_size bs = AES.block_size
tmpsalt = in_file.read(bs) b_tmpsalt = in_file.read(bs)
salt = tmpsalt[len(b'Salted__'):] b_salt = b_tmpsalt[len(b'Salted__'):]
key, iv = self.aes_derive_key_and_iv(password, salt, key_length, bs) b_key, b_iv = self._aes_derive_key_and_iv(b_password, b_salt, key_length, bs)
cipher = AES.new(key, AES.MODE_CBC, iv) cipher = AES.new(b_key, AES.MODE_CBC, b_iv)
next_chunk = b'' b_next_chunk = b''
finished = False finished = False
while not finished: while not finished:
chunk, next_chunk = next_chunk, cipher.decrypt(in_file.read(1024 * bs)) b_chunk, b_next_chunk = b_next_chunk, cipher.decrypt(in_file.read(1024 * bs))
if len(next_chunk) == 0: if len(b_next_chunk) == 0:
if PY3: if PY3:
padding_length = chunk[-1] padding_length = b_chunk[-1]
else: else:
padding_length = ord(chunk[-1]) padding_length = ord(b_chunk[-1])
chunk = chunk[:-padding_length] b_chunk = b_chunk[:-padding_length]
finished = True finished = True
out_file.write(chunk) out_file.write(b_chunk)
out_file.flush() out_file.flush()
# reset the stream pointer to the beginning # reset the stream pointer to the beginning
out_file.seek(0) out_file.seek(0)
out_data = out_file.read() b_out_data = out_file.read()
out_file.close() out_file.close()
# split out sha and verify decryption # split out sha and verify decryption
split_data = out_data.split(b"\n", 1) b_split_data = b_out_data.split(b"\n", 1)
this_sha = split_data[0] b_this_sha = b_split_data[0]
this_data = split_data[1] b_plaintext = b_split_data[1]
test_sha = to_bytes(sha256(this_data).hexdigest()) b_test_sha = to_bytes(sha256(b_plaintext).hexdigest())
if this_sha != test_sha: if b_this_sha != b_test_sha:
raise AnsibleError("Decryption failed") raise AnsibleError("Decryption failed")
return this_data return b_plaintext
class VaultAES256: class VaultAES256:
@ -678,17 +666,19 @@ class VaultAES256:
check_prereqs() check_prereqs()
def create_key(self, password, salt, keylength, ivlength): @staticmethod
def _create_key(b_password, b_salt, keylength, ivlength):
hash_function = SHA256 hash_function = SHA256
# make two keys and one iv # make two keys and one iv
pbkdf2_prf = lambda p, s: HMAC.new(p, s, hash_function).digest() pbkdf2_prf = lambda p, s: HMAC.new(p, s, hash_function).digest()
derivedkey = PBKDF2(password, salt, dkLen=(2 * keylength) + ivlength, b_derivedkey = PBKDF2(b_password, b_salt, dkLen=(2 * keylength) + ivlength,
count=10000, prf=pbkdf2_prf) count=10000, prf=pbkdf2_prf)
return derivedkey return b_derivedkey
def gen_key_initctr(self, password, salt): @classmethod
def _gen_key_initctr(cls, b_password, b_salt):
# 16 for AES 128, 32 for AES256 # 16 for AES 128, 32 for AES256
keylength = 32 keylength = 32
@ -700,79 +690,80 @@ class VaultAES256:
kdf = PBKDF2HMAC( kdf = PBKDF2HMAC(
algorithm=c_SHA256(), algorithm=c_SHA256(),
length=2 * keylength + ivlength, length=2 * keylength + ivlength,
salt=salt, salt=b_salt,
iterations=10000, iterations=10000,
backend=backend) backend=backend)
derivedkey = kdf.derive(password) b_derivedkey = kdf.derive(b_password)
else: else:
derivedkey = self.create_key(password, salt, keylength, ivlength) b_derivedkey = cls._create_key(b_password, b_salt, keylength, ivlength)
key1 = derivedkey[:keylength] b_key1 = b_derivedkey[:keylength]
key2 = derivedkey[keylength:(keylength * 2)] b_key2 = b_derivedkey[keylength:(keylength * 2)]
iv = derivedkey[(keylength * 2):(keylength * 2) + ivlength] b_iv = b_derivedkey[(keylength * 2):(keylength * 2) + ivlength]
return key1, key2, hexlify(iv) return b_key1, b_key2, hexlify(b_iv)
def encrypt(self, data, password): def encrypt(self, b_plaintext, b_password):
salt = os.urandom(32) b_salt = os.urandom(32)
key1, key2, iv = self.gen_key_initctr(password, salt) b_key1, b_key2, b_iv = self._gen_key_initctr(b_password, b_salt)
# PKCS#7 PAD DATA http://tools.ietf.org/html/rfc5652#section-6.3 # PKCS#7 PAD DATA http://tools.ietf.org/html/rfc5652#section-6.3
bs = AES.block_size bs = AES.block_size
padding_length = (bs - len(data) % bs) or bs padding_length = (bs - len(b_plaintext) % bs) or bs
data += to_bytes(padding_length * chr(padding_length), encoding='ascii', errors='strict') b_plaintext += to_bytes(padding_length * chr(padding_length), encoding='ascii', errors='strict')
# COUNTER.new PARAMETERS # COUNTER.new PARAMETERS
# 1) nbits (integer) - Length of the counter, in bits. # 1) nbits (integer) - Length of the counter, in bits.
# 2) initial_value (integer) - initial value of the counter. "iv" from gen_key_initctr # 2) initial_value (integer) - initial value of the counter. "iv" from _gen_key_initctr
ctr = Counter.new(128, initial_value=int(iv, 16)) ctr = Counter.new(128, initial_value=int(b_iv, 16))
# AES.new PARAMETERS # AES.new PARAMETERS
# 1) AES key, must be either 16, 24, or 32 bytes long -- "key" from gen_key_initctr # 1) AES key, must be either 16, 24, or 32 bytes long -- "key" from _gen_key_initctr
# 2) MODE_CTR, is the recommended mode # 2) MODE_CTR, is the recommended mode
# 3) counter=<CounterObject> # 3) counter=<CounterObject>
cipher = AES.new(key1, AES.MODE_CTR, counter=ctr) cipher = AES.new(b_key1, AES.MODE_CTR, counter=ctr)
# ENCRYPT PADDED DATA # ENCRYPT PADDED DATA
cryptedData = cipher.encrypt(data) b_ciphertext = cipher.encrypt(b_plaintext)
# COMBINE SALT, DIGEST AND DATA # COMBINE SALT, DIGEST AND DATA
hmac = HMAC.new(key2, cryptedData, SHA256) hmac = HMAC.new(b_key2, b_ciphertext, SHA256)
message = b'\n'.join([hexlify(salt), to_bytes(hmac.hexdigest()), hexlify(cryptedData)]) b_vaulttext = b'\n'.join([hexlify(b_salt), to_bytes(hmac.hexdigest()), hexlify(b_ciphertext)])
message = hexlify(message) b_vaulttext = hexlify(b_vaulttext)
return message return b_vaulttext
def decrypt(self, data, password): def decrypt(self, b_vaulttext, b_password):
# SPLIT SALT, DIGEST, AND DATA # SPLIT SALT, DIGEST, AND DATA
data = unhexlify(data) b_vaulttext = unhexlify(b_vaulttext)
salt, cryptedHmac, cryptedData = data.split(b"\n", 2) b_salt, b_cryptedHmac, b_ciphertext = b_vaulttext.split(b"\n", 2)
salt = unhexlify(salt) b_salt = unhexlify(b_salt)
cryptedData = unhexlify(cryptedData) b_ciphertext = unhexlify(b_ciphertext)
key1, key2, iv = self.gen_key_initctr(password, salt) b_key1, b_key2, b_iv = self._gen_key_initctr(b_password, b_salt)
# EXIT EARLY IF DIGEST DOESN'T MATCH # EXIT EARLY IF DIGEST DOESN'T MATCH
hmacDecrypt = HMAC.new(key2, cryptedData, SHA256) hmacDecrypt = HMAC.new(b_key2, b_ciphertext, SHA256)
if not self.is_equal(cryptedHmac, to_bytes(hmacDecrypt.hexdigest())): if not self._is_equal(b_cryptedHmac, to_bytes(hmacDecrypt.hexdigest())):
return None return None
# SET THE COUNTER AND THE CIPHER # SET THE COUNTER AND THE CIPHER
ctr = Counter.new(128, initial_value=int(iv, 16)) ctr = Counter.new(128, initial_value=int(b_iv, 16))
cipher = AES.new(key1, AES.MODE_CTR, counter=ctr) cipher = AES.new(b_key1, AES.MODE_CTR, counter=ctr)
# DECRYPT PADDED DATA # DECRYPT PADDED DATA
decryptedData = cipher.decrypt(cryptedData) b_plaintext = cipher.decrypt(b_ciphertext)
# UNPAD DATA # UNPAD DATA
try: if PY3:
padding_length = ord(decryptedData[-1]) padding_length = b_plaintext[-1]
except TypeError: else:
padding_length = decryptedData[-1] padding_length = ord(b_plaintext[-1])
decryptedData = decryptedData[:-padding_length] b_plaintext = b_plaintext[:-padding_length]
return decryptedData return b_plaintext
def is_equal(self, a, b): @staticmethod
def _is_equal(b_a, b_b):
""" """
Comparing 2 byte arrrays in constant time Comparing 2 byte arrrays in constant time
to avoid timing attacks. to avoid timing attacks.
@ -780,16 +771,19 @@ class VaultAES256:
It would be nice if there was a library for this but It would be nice if there was a library for this but
hey. hey.
""" """
if not (isinstance(b_a, binary_type) and isinstance(b_b, binary_type)):
raise TypeError('_is_equal can only be used to compare two byte strings')
# http://codahale.com/a-lesson-in-timing-attacks/ # http://codahale.com/a-lesson-in-timing-attacks/
if len(a) != len(b): if len(b_a) != len(b_b):
return False return False
result = 0 result = 0
for x, y in zip(a, b): for b_x, b_y in zip(b_a, b_b):
if PY3: if PY3:
result |= x ^ y result |= b_x ^ b_y
else: else:
result |= ord(x) ^ ord(y) result |= ord(b_x) ^ ord(b_y)
return result == 0 return result == 0

4
test/units/cli/test_galaxy.py
Unescape Escape View File

@ -24,11 +24,9 @@ import shutil
import tarfile import tarfile
import tempfile import tempfile
from mock import patch
from ansible.compat.six import PY3 from ansible.compat.six import PY3
from ansible.compat.tests import unittest from ansible.compat.tests import unittest
from mock import patch, call from ansible.compat.tests.mock import call, patch
import ansible import ansible
from ansible.errors import AnsibleError, AnsibleOptionsError from ansible.errors import AnsibleError, AnsibleOptionsError

3
test/units/module_utils/test_database.py
Unescape Escape View File

@ -1,5 +1,4 @@
import collections import collections
import mock
import os import os
import re import re
@ -24,6 +23,8 @@ except ImportError:
else: excName = str(expected) else: excName = str(expected)
raise AssertionError("%s not raised" % excName) raise AssertionError("%s not raised" % excName)
from ansible.compat.tests import mock
from ansible.module_utils.database import ( from ansible.module_utils.database import (
pg_quote_identifier, pg_quote_identifier,
SQLParseError, SQLParseError,

4
test/units/modules/core/test_apt.py
Unescape Escape View File

@ -1,9 +1,9 @@
import collections import collections
import mock
import os import os
import sys import sys
from ansible.compat.tests import unittest
from ansible.compat.tests import mock
from ansible.compat.tests import unittest
try: try:
from ansible.modules.core.packaging.os.apt import ( from ansible.modules.core.packaging.os.apt import (

345
test/units/parsing/vault/test_vault.py
Unescape Escape View File

@ -1,5 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com> # (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
# (c) 2016, Toshio Kuratomi <tkuratomi@ansible.com>
# #
# This file is part of Ansible # This file is part of Ansible
# #
@ -34,7 +35,7 @@ from ansible.compat.tests import unittest
from ansible import errors from ansible import errors
from ansible.parsing.vault import VaultLib from ansible.parsing.vault import VaultLib
from ansible.parsing import vault from ansible.parsing import vault
from ansible.module_utils._text import to_bytes from ansible.module_utils._text import to_bytes, to_text
# Counter import fails for 2.0.1, requires >= 2.6.1 from pip # Counter import fails for 2.0.1, requires >= 2.6.1 from pip
@ -60,241 +61,254 @@ except ImportError:
class TestVaultIsEncrypted(unittest.TestCase): class TestVaultIsEncrypted(unittest.TestCase):
def test_utf8_not_encrypted(self): def test_bytes_not_encrypted(self):
b_data = "foobar".encode('utf8') b_data = b"foobar"
self.assertFalse(vault.is_encrypted(b_data)) self.assertFalse(vault.is_encrypted(b_data))
def test_utf8_encrypted(self): def test_bytes_encrypted(self):
data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible") b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
b_data = data.encode('utf8')
self.assertTrue(vault.is_encrypted(b_data)) self.assertTrue(vault.is_encrypted(b_data))
def test_bytes_not_encrypted(self): def test_text_not_encrypted(self):
b_data = b"foobar" b_data = to_text(b"foobar")
self.assertFalse(vault.is_encrypted(b_data)) self.assertFalse(vault.is_encrypted(b_data))
def test_bytes_encrypted(self): def test_text_encrypted(self):
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" + hexlify(b"ansible") b_data = to_text(b"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible"))
self.assertTrue(vault.is_encrypted(b_data)) self.assertTrue(vault.is_encrypted(b_data))
def test_unicode_not_encrypted_py3(self): def test_invalid_text_not_ascii(self):
if not six.PY3: data = u"$ANSIBLE_VAULT;9.9;TEST\n%s"% u"ァ ア ィ イ ゥ ウ ェ エ ォ オ カ ガ キ ギ ク グ ケ "
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)) self.assertFalse(vault.is_encrypted(data))
def test_unicode_is_encrypted_py3(self): def test_invalid_bytes_not_ascii(self):
if not six.PY3: data = u"$ANSIBLE_VAULT;9.9;TEST\n%s"% u"ァ ア ィ イ ゥ ウ ェ エ ォ オ カ ガ キ ギ ク グ ケ "
raise SkipTest() b_data = to_bytes(data, encoding='utf-8')
data = "$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible") self.assertFalse(vault.is_encrypted(b_data))
# 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): class TestVaultIsEncryptedFile(unittest.TestCase):
def test_utf8_not_encrypted(self): def test_binary_file_handle_not_encrypted(self):
b_data = "foobar".encode('utf8') b_data = b"foobar"
b_data_fo = io.BytesIO(b_data) b_data_fo = io.BytesIO(b_data)
self.assertFalse(vault.is_encrypted_file(b_data_fo)) self.assertFalse(vault.is_encrypted_file(b_data_fo))
def test_utf8_encrypted(self): def test_text_file_handle_not_encrypted(self):
data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible") data = u"foobar"
b_data = data.encode('utf8') data_fo = io.StringIO(data)
self.assertFalse(vault.is_encrypted_file(data_fo))
def test_binary_file_handle_encrypted(self):
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
b_data_fo = io.BytesIO(b_data) b_data_fo = io.BytesIO(b_data)
self.assertTrue(vault.is_encrypted_file(b_data_fo)) self.assertTrue(vault.is_encrypted_file(b_data_fo))
def test_bytes_not_encrypted(self): def test_text_file_handle_encrypted(self):
b_data = b"foobar" data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % to_text(hexlify(b"ansible"))
data_fo = io.StringIO(data)
self.assertTrue(vault.is_encrypted_file(data_fo))
def test_binary_file_handle_invalid(self):
data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % u"ァ ア ィ イ ゥ ウ ェ エ ォ オ カ ガ キ ギ ク グ ケ "
b_data = to_bytes(data)
b_data_fo = io.BytesIO(b_data) b_data_fo = io.BytesIO(b_data)
self.assertFalse(vault.is_encrypted_file(b_data_fo)) self.assertFalse(vault.is_encrypted_file(b_data_fo))
def test_bytes_encrypted(self): def test_text_file_handle_invalid(self):
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" + hexlify(b"ansible") data = u"$ANSIBLE_VAULT;9.9;TEST\n%s" % u"ァ ア ィ イ ゥ ウ ェ エ ォ オ カ ガ キ ギ ク グ ケ "
data_fo = io.StringIO(data)
self.assertFalse(vault.is_encrypted_file(data_fo))
def test_file_already_read_from_finds_header(self):
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible\ntesting\nfile pos")
b_data_fo = io.BytesIO(b_data) b_data_fo = io.BytesIO(b_data)
b_data_fo.read(42) # Arbitrary number
self.assertTrue(vault.is_encrypted_file(b_data_fo)) self.assertTrue(vault.is_encrypted_file(b_data_fo))
def test_file_already_read_from_saves_file_pos(self):
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible\ntesting\nfile pos")
b_data_fo = io.BytesIO(b_data)
b_data_fo.read(69) # Arbitrary number
vault.is_encrypted_file(b_data_fo)
self.assertEqual(b_data_fo.tell(), 69)
def test_file_with_offset(self):
b_data = b"JUNK$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible\ntesting\nfile pos")
b_data_fo = io.BytesIO(b_data)
self.assertTrue(vault.is_encrypted_file(b_data_fo, start_pos=4))
def test_file_with_count(self):
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible\ntesting\nfile pos")
vault_length = len(b_data)
b_data = b_data + u'ァ ア'.encode('utf-8')
b_data_fo = io.BytesIO(b_data)
self.assertTrue(vault.is_encrypted_file(b_data_fo, count=vault_length))
def test_file_with_offset_and_count(self):
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible\ntesting\nfile pos")
vault_length = len(b_data)
b_data = b'JUNK' + b_data + u'ァ ア'.encode('utf-8')
b_data_fo = io.BytesIO(b_data)
self.assertTrue(vault.is_encrypted_file(b_data_fo, start_pos=4, count=vault_length))
class TestVaultCipherAes256(unittest.TestCase): class TestVaultCipherAes256(unittest.TestCase):
def setUp(self):
self.vault_cipher = vault.VaultAES256()
def test(self): def test(self):
vault_cipher = vault.VaultAES256() self.assertIsInstance(self.vault_cipher, vault.VaultAES256)
self.assertIsInstance(vault_cipher, vault.VaultAES256)
# TODO: tag these as slow tests # TODO: tag these as slow tests
def test_create_key(self): def test_create_key(self):
vault_cipher = vault.VaultAES256() b_password = b'hunter42'
password = 'hunter42'
b_salt = os.urandom(32) b_salt = os.urandom(32)
b_key = vault_cipher.create_key(password=password, salt=b_salt, keylength=32, ivlength=16) b_key = self.vault_cipher._create_key(b_password, b_salt, keylength=32, ivlength=16)
self.assertIsInstance(b_key, six.binary_type) self.assertIsInstance(b_key, six.binary_type)
def test_create_key_known(self): def test_create_key_known(self):
vault_cipher = vault.VaultAES256() b_password = b'hunter42'
password = 'hunter42'
# A fixed salt # A fixed salt
b_salt = b'q' * 32 # q is the most random letter. 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) b_key = self.vault_cipher._create_key(b_password, b_salt, keylength=32, ivlength=16)
self.assertIsInstance(b_key, six.binary_type) self.assertIsInstance(b_key, six.binary_type)
# verify we get the same answer # 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 # 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 # 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) b_key_2 = self.vault_cipher._create_key(b_password, b_salt, keylength=32, ivlength=16)
self.assertIsInstance(b_key, six.binary_type) self.assertIsInstance(b_key, six.binary_type)
self.assertEqual(b_key, b_key_2) self.assertEqual(b_key, b_key_2)
def test_is_equal_is_equal(self): def test_is_equal_is_equal(self):
vault_cipher = vault.VaultAES256() self.assertTrue(self.vault_cipher._is_equal(b'abcdefghijklmnopqrstuvwxyz', b'abcdefghijklmnopqrstuvwxyz'))
res = vault_cipher.is_equal(b'abcdefghijklmnopqrstuvwxyz', b'abcdefghijklmnopqrstuvwxyz')
self.assertTrue(res)
def test_is_equal_unequal_length(self): def test_is_equal_unequal_length(self):
vault_cipher = vault.VaultAES256() self.assertFalse(self.vault_cipher._is_equal(b'abcdefghijklmnopqrstuvwxyz', b'abcdefghijklmnopqrstuvwx and sometimes y'))
res = vault_cipher.is_equal(b'abcdefghijklmnopqrstuvwxyz', b'abcdefghijklmnopqrstuvwx and sometimes y')
self.assertFalse(res)
def test_is_equal_not_equal(self): def test_is_equal_not_equal(self):
vault_cipher = vault.VaultAES256() self.assertFalse(self.vault_cipher._is_equal(b'abcdefghijklmnopqrstuvwxyz', b'AbcdefghijKlmnopQrstuvwxZ'))
res = vault_cipher.is_equal(b'abcdefghijklmnopqrstuvwxyz', b'AbcdefghijKlmnopQrstuvwxZ')
self.assertFalse(res)
def test_is_equal_empty(self): def test_is_equal_empty(self):
vault_cipher = vault.VaultAES256() self.assertTrue(self.vault_cipher._is_equal(b'', b''))
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): def test_is_equal_non_ascii_equal(self):
if not six.PY2: utf8_data = to_bytes(u'私はガラスを食べられます。それは私を傷つけません。')
raise SkipTest self.assertTrue(self.vault_cipher._is_equal(utf8_data, utf8_data))
vault_cipher = vault.VaultAES256()
res = vault_cipher.is_equal(u'私はガラスを食べられます。それは私を傷つけません。',
u'私はガラスを食べられます。それは私を傷つけません。')
self.assertTrue(res)
def test_is_equal_unicode_different(self): def test_is_equal_non_ascii_unequal(self):
vault_cipher = vault.VaultAES256() utf8_data = to_bytes(u'私はガラスを食べられます。それは私を傷つけません。')
res = vault_cipher.is_equal(u'私はガラスを食べられます。それは私を傷つけません。', utf8_data2 = to_bytes(u'Pot să mănânc sticlă și ea nu mă rănește.')
u'Pot să mănânc sticlă și ea nu mă rănește.')
self.assertFalse(res)
# Test for the len optimization path
self.assertFalse(self.vault_cipher._is_equal(utf8_data, utf8_data2))
# Test for the slower, char by char comparison path
self.assertFalse(self.vault_cipher._is_equal(utf8_data, utf8_data[:-1] + b'P'))
class TestVaultLib(unittest.TestCase): def test_is_equal_non_bytes(self):
""" Anything not a byte string should raise a TypeError """
self.assertRaises(TypeError, self.vault_cipher._is_equal, u"One fish", b"two fish")
self.assertRaises(TypeError, self.vault_cipher._is_equal, b"One fish", u"two fish")
self.assertRaises(TypeError, self.vault_cipher._is_equal, 1, b"red fish")
self.assertRaises(TypeError, self.vault_cipher._is_equal, b"blue fish", 2)
def test_methods_exist(self):
v = VaultLib('ansible') class TestVaultLib(unittest.TestCase):
slots = ['is_encrypted', def setUp(self):
'encrypt', self.v = VaultLib('test-vault-password')
'decrypt',
'_format_output',
'_split_header',]
for slot in slots:
assert hasattr(v, slot), "VaultLib is missing the %s method" % slot
def test_encrypt(self): def test_encrypt(self):
v = VaultLib(password='the_unit_test_password') plaintext = u'Some text to encrypt in a café'
plaintext = u'Some text to encrypt.' b_vaulttext = self.v.encrypt(plaintext)
ciphertext = v.encrypt(plaintext)
self.assertIsInstance(ciphertext, (bytes, str)) self.assertIsInstance(b_vaulttext, six.binary_type)
# TODO: assert something...
def test_is_encrypted(self): b_header = b'$ANSIBLE_VAULT;1.1;AES256\n'
v = VaultLib(None) self.assertEqual(b_vaulttext[:len(b_header)], b_header)
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): def test_encrypt_bytes(self):
v = VaultLib(None)
assert not v.is_encrypted(b"foobar"), "encryption check on plaintext failed" plaintext = to_bytes(u'Some text to encrypt in a café')
data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" + hexlify(b"ansible") b_vaulttext = self.v.encrypt(plaintext)
assert v.is_encrypted(data), "encryption check on headered text failed"
self.assertIsInstance(b_vaulttext, six.binary_type)
b_header = b'$ANSIBLE_VAULT;1.1;AES256\n'
self.assertEqual(b_vaulttext[:len(b_header)], b_header)
def test_is_encrypted(self):
self.assertFalse(self.v.is_encrypted(b"foobar"), msg="encryption check on plaintext yielded false positive")
b_data = b"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
self.assertTrue(self.v.is_encrypted(b_data), msg="encryption check on headered text failed")
def test_format_output(self): def test_format_output(self):
v = VaultLib('ansible') self.v.cipher_name = "TEST"
v.cipher_name = "TEST" b_ciphertext = b"ansible"
sensitive_data = b"ansible" b_vaulttext = self.v._format_output(b_ciphertext)
data = v._format_output(sensitive_data) b_lines = b_vaulttext.split(b'\n')
lines = data.split(b'\n') self.assertGreater(len(b_lines), 1, msg="failed to properly add header")
assert len(lines) > 1, "failed to properly add header"
header = to_bytes(lines[0]) b_header = b_lines[0]
assert header.endswith(b';TEST'), "header does end with cipher name" self.assertTrue(b_header.endswith(b';TEST'), msg="header does not end with cipher name")
header_parts = header.split(b';')
assert len(header_parts) == 3, "header has the wrong number of parts" b_header_parts = b_header.split(b';')
assert header_parts[0] == b'$ANSIBLE_VAULT', "header does not start with $ANSIBLE_VAULT" self.assertEqual(len(b_header_parts), 3, msg="header has the wrong number of parts")
assert header_parts[1] == v.b_version, "header version is incorrect" self.assertEqual(b_header_parts[0], b'$ANSIBLE_VAULT', msg="header does not start with $ANSIBLE_VAULT")
assert header_parts[2] == b'TEST', "header does end with cipher name" self.assertEqual(b_header_parts[1], self.v.b_version, msg="header version is incorrect")
self.assertEqual(b_header_parts[2], b'TEST', msg="header does not end with cipher name")
def test_split_header(self): def test_split_header(self):
v = VaultLib('ansible') b_vaulttext = b"$ANSIBLE_VAULT;9.9;TEST\nansible"
data = b"$ANSIBLE_VAULT;9.9;TEST\nansible" b_ciphertext = self.v._split_header(b_vaulttext)
rdata = v._split_header(data) b_lines = b_ciphertext.split(b'\n')
lines = rdata.split(b'\n') self.assertEqual(b_lines[0], b"ansible", msg="Payload was not properly split from the header")
assert lines[0] == b"ansible" self.assertEqual(self.v.cipher_name, u'TEST', msg="cipher name was not properly set")
assert v.cipher_name == 'TEST', "cipher name was not set" self.assertEqual(self.v.b_version, b"9.9", msg="version was not properly set")
assert v.b_version == b"9.9"
def test_encrypt_decrypt_aes(self): def test_encrypt_decrypt_aes(self):
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2: if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest raise SkipTest
v = VaultLib('ansible') self.v.cipher_name = u'AES'
v.cipher_name = u'AES' self.v.b_password = b'ansible'
# AES encryption code has been removed, so this is old output for # AES encryption code has been removed, so this is old output for
# AES-encrypted 'foobar' with password 'ansible'. # AES-encrypted 'foobar' with password 'ansible'.
enc_data = b'$ANSIBLE_VAULT;1.1;AES\n53616c7465645f5fc107ce1ef4d7b455e038a13b053225776458052f8f8f332d554809d3f150bfa3\nfe3db930508b65e0ff5947e4386b79af8ab094017629590ef6ba486814cf70f8e4ab0ed0c7d2587e\n786a5a15efeb787e1958cbdd480d076c\n' b_vaulttext = b'''$ANSIBLE_VAULT;1.1;AES
dec_data = v.decrypt(enc_data) 53616c7465645f5fc107ce1ef4d7b455e038a13b053225776458052f8f8f332d554809d3f150bfa3
assert dec_data == b"foobar", "decryption failed" fe3db930508b65e0ff5947e4386b79af8ab094017629590ef6ba486814cf70f8e4ab0ed0c7d2587e
786a5a15efeb787e1958cbdd480d076c
'''
b_plaintext = self.v.decrypt(b_vaulttext)
self.assertEqual(b_plaintext, b"foobar", msg="decryption failed")
def test_encrypt_decrypt_aes256(self): def test_encrypt_decrypt_aes256(self):
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2: if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest raise SkipTest
v = VaultLib('ansible') self.v.cipher_name = u'AES256'
v.cipher_name = 'AES256' plaintext = u"foobar"
plaintext = "foobar" b_vaulttext = self.v.encrypt(plaintext)
enc_data = v.encrypt(plaintext) b_plaintext = self.v.decrypt(b_vaulttext)
dec_data = v.decrypt(enc_data) self.assertNotEqual(b_vaulttext, b"foobar", msg="encryption failed")
assert enc_data != b"foobar", "encryption failed" self.assertEqual(b_plaintext, b"foobar", msg="decryption failed")
assert dec_data == b"foobar", "decryption failed"
def test_encrypt_decrypt_aes256_existing_vault(self): def test_encrypt_decrypt_aes256_existing_vault(self):
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2: if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest raise SkipTest
v = VaultLib('test-vault-password') self.v.cipher_name = u'AES256'
v.cipher_name = 'AES256' b_orig_plaintext = b"Setec Astronomy"
plaintext = b"Setec Astronomy" vaulttext = u'''$ANSIBLE_VAULT;1.1;AES256
enc_data = '''$ANSIBLE_VAULT;1.1;AES256
33363965326261303234626463623963633531343539616138316433353830356566396130353436 33363965326261303234626463623963633531343539616138316433353830356566396130353436
3562643163366231316662386565383735653432386435610a306664636137376132643732393835 3562643163366231316662386565383735653432386435610a306664636137376132643732393835
63383038383730306639353234326630666539346233376330303938323639306661313032396437 63383038383730306639353234326630666539346233376330303938323639306661313032396437
6233623062366136310a633866373936313238333730653739323461656662303864663666653563 6233623062366136310a633866373936313238333730653739323461656662303864663666653563
3138''' 3138'''
dec_data = v.decrypt(enc_data) b_plaintext = self.v.decrypt(vaulttext)
assert dec_data == plaintext, "decryption failed" self.assertEqual(b_plaintext, b_plaintext, msg="decryption failed")
b_vaulttext = to_bytes(vaulttext, encoding='ascii', errors='strict')
b_plaintext = self.v.decrypt(b_vaulttext)
self.assertEqual(b_plaintext, b_orig_plaintext, msg="decryption failed")
def test_encrypt_decrypt_aes256_bad_hmac(self): def test_encrypt_decrypt_aes256_bad_hmac(self):
# FIXME This test isn't working quite yet. # FIXME This test isn't working quite yet.
@ -302,8 +316,7 @@ class TestVaultLib(unittest.TestCase):
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2: if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest raise SkipTest
v = VaultLib('test-vault-password') self.v.cipher_name = 'AES256'
v.cipher_name = 'AES256'
# plaintext = "Setec Astronomy" # plaintext = "Setec Astronomy"
enc_data = '''$ANSIBLE_VAULT;1.1;AES256 enc_data = '''$ANSIBLE_VAULT;1.1;AES256
33363965326261303234626463623963633531343539616138316433353830356566396130353436 33363965326261303234626463623963633531343539616138316433353830356566396130353436
@ -312,7 +325,7 @@ class TestVaultLib(unittest.TestCase):
6233623062366136310a633866373936313238333730653739323461656662303864663666653563 6233623062366136310a633866373936313238333730653739323461656662303864663666653563
3138''' 3138'''
b_data = to_bytes(enc_data, errors='strict', encoding='utf-8') b_data = to_bytes(enc_data, errors='strict', encoding='utf-8')
b_data = v._split_header(b_data) b_data = self.v._split_header(b_data)
foo = binascii.unhexlify(b_data) foo = binascii.unhexlify(b_data)
lines = foo.splitlines() lines = foo.splitlines()
# line 0 is salt, line 1 is hmac, line 2+ is ciphertext # line 0 is salt, line 1 is hmac, line 2+ is ciphertext
@ -331,31 +344,33 @@ class TestVaultLib(unittest.TestCase):
b_ciphertext_data = binascii.hexlify(b_ciphertext) b_ciphertext_data = binascii.hexlify(b_ciphertext)
b_payload = b'\n'.join([b_salt, b_hmac, b_ciphertext_data]) b_payload = b'\n'.join([b_salt, b_hmac, b_ciphertext_data])
# reformat # reformat
b_invalid_ciphertext = v._format_output(b_payload) b_invalid_ciphertext = self.v._format_output(b_payload)
# assert we throw an error # assert we throw an error
v.decrypt(b_invalid_ciphertext) self.v.decrypt(b_invalid_ciphertext)
def test_encrypt_encrypted(self): def test_encrypt_encrypted(self):
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2: if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest raise SkipTest
v = VaultLib('ansible') self.v.cipher_name = u'AES'
v.cipher_name = 'AES' b_vaulttext = b"$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(b"ansible")
data = "$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify(six.b("ansible")) vaulttext = to_text(b_vaulttext, errors='strict')
self.assertRaises(errors.AnsibleError, v.encrypt, data,) self.assertRaises(errors.AnsibleError, self.v.encrypt, b_vaulttext)
self.assertRaises(errors.AnsibleError, self.v.encrypt, vaulttext)
def test_decrypt_decrypted(self): def test_decrypt_decrypted(self):
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2: if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest raise SkipTest
v = VaultLib('ansible') plaintext = u"ansible"
data = "ansible" self.assertRaises(errors.AnsibleError, self.v.decrypt, plaintext)
self.assertRaises(errors.AnsibleError, v.decrypt, data)
b_plaintext = b"ansible"
self.assertRaises(errors.AnsibleError, self.v.decrypt, b_plaintext)
def test_cipher_not_set(self): def test_cipher_not_set(self):
# not setting the cipher should default to AES256 # not setting the cipher should default to AES256
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2: if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest raise SkipTest
v = VaultLib('ansible') plaintext = u"ansible"
data = "ansible" self.v.encrypt(plaintext)
v.encrypt(data) self.assertEquals(self.v.cipher_name, "AES256")
self.assertEquals(v.cipher_name, "AES256")

14
test/units/parsing/vault/test_vault_editor.py
Unescape Escape View File

@ -104,12 +104,8 @@ class TestVaultEditor(unittest.TestCase):
self.assertTrue(os.path.exists(tmp_file.name)) self.assertTrue(os.path.exists(tmp_file.name))
@unittest.skipIf(sys.version_info[0] >= 3, "VaultAES still needs to be ported to Python 3")
def test_decrypt_1_0(self): def test_decrypt_1_0(self):
""" # Skip testing decrypting 1.0 files if we don't have access to AES, KDF or Counter.
Skip testing decrypting 1.0 files if we don't have access to AES, KDF or
Counter, or we are running on python3 since VaultAES hasn't been backported.
"""
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2: if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest raise SkipTest
@ -163,12 +159,8 @@ class TestVaultEditor(unittest.TestCase):
assert error_hit is False, "error decrypting 1.0 file" assert error_hit is False, "error decrypting 1.0 file"
assert fdata.strip() == "foo", "incorrect decryption of 1.0 file: %s" % fdata.strip() assert fdata.strip() == "foo", "incorrect decryption of 1.0 file: %s" % fdata.strip()
@unittest.skipIf(sys.version_info[0] >= 3, "VaultAES still needs to be ported to Python 3")
def test_rekey_migration(self): def test_rekey_migration(self):
""" # Skip testing rekeying files if we don't have access to AES, KDF or Counter.
Skip testing rekeying files if we don't have access to AES, KDF or
Counter, or we are running on python3 since VaultAES hasn't been backported.
"""
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2: if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
raise SkipTest raise SkipTest
@ -205,4 +197,4 @@ class TestVaultEditor(unittest.TestCase):
assert vl.cipher_name == "AES256", "wrong cipher name set after rekey: %s" % vl.cipher_name assert vl.cipher_name == "AES256", "wrong cipher name set after rekey: %s" % vl.cipher_name
assert error_hit is False, "error decrypting migrated 1.0 file" assert error_hit is False, "error decrypting migrated 1.0 file"
assert dec_data.strip() == "foo", "incorrect decryption of rekeyed/migrated file: %s" % dec_data assert dec_data.strip() == b"foo", "incorrect decryption of rekeyed/migrated file: %s" % dec_data

3
test/units/plugins/connections/test_connection.py
Unescape Escape View File

@ -21,8 +21,7 @@ __metaclass__ = type
from io import StringIO from io import StringIO
import mock from ansible.compat.tests import mock
from ansible.compat.tests import unittest from ansible.compat.tests import unittest
from ansible.errors import AnsibleError from ansible.errors import AnsibleError
from ansible.playbook.play_context import PlayContext from ansible.playbook.play_context import PlayContext

Write Preview
Loading…
Cancel
Save