Implement new default cipher class AES256
This commit is contained in:
James Tanner
James Cammarata
parent
ba0fec4f42
commit
0d6f6ad282
6 changed files with 410 additions and 51 deletions
|
|
@ -52,7 +52,7 @@ def build_option_parser(action):
|
|||
sys.exit()
|
||||
|
||||
# options for all actions
|
||||
#parser.add_option('-c', '--cipher', dest='cipher', default="AES", help="cipher to use")
|
||||
#parser.add_option('-c', '--cipher', dest='cipher', default="AES256", help="cipher to use")
|
||||
parser.add_option('--debug', dest='debug', action="store_true", help="debug")
|
||||
parser.add_option('--vault-password-file', dest='password_file',
|
||||
help="vault password file")
|
||||
|
|
@ -119,7 +119,7 @@ def execute_create(args, options, parser):
|
|||
else:
|
||||
password = _read_password(options.password_file)
|
||||
|
||||
cipher = 'AES'
|
||||
cipher = 'AES256'
|
||||
if hasattr(options, 'cipher'):
|
||||
cipher = options.cipher
|
||||
|
||||
|
|
@ -133,7 +133,7 @@ def execute_decrypt(args, options, parser):
|
|||
else:
|
||||
password = _read_password(options.password_file)
|
||||
|
||||
cipher = 'AES'
|
||||
cipher = 'AES256'
|
||||
if hasattr(options, 'cipher'):
|
||||
cipher = options.cipher
|
||||
|
||||
|
|
@ -169,7 +169,7 @@ def execute_encrypt(args, options, parser):
|
|||
else:
|
||||
password = _read_password(options.password_file)
|
||||
|
||||
cipher = 'AES'
|
||||
cipher = 'AES256'
|
||||
if hasattr(options, 'cipher'):
|
||||
cipher = options.cipher
|
||||
|
||||
|
|
|
|||
|
|
@ -30,6 +30,22 @@ from binascii import hexlify
|
|||
from binascii import unhexlify
|
||||
from ansible import constants as C
|
||||
|
||||
from Crypto.Hash import SHA256, HMAC
|
||||
|
||||
# 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
|
||||
|
|
@ -37,15 +53,17 @@ try:
|
|||
except ImportError:
|
||||
HAS_AES = False
|
||||
|
||||
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: rpm -e --nodeps python-crypto; pip install pycrypto"
|
||||
|
||||
HEADER='$ANSIBLE_VAULT'
|
||||
CIPHER_WHITELIST=['AES']
|
||||
CIPHER_WHITELIST=['AES', 'AES256']
|
||||
|
||||
class VaultLib(object):
|
||||
|
||||
def __init__(self, password):
|
||||
self.password = password
|
||||
self.cipher_name = None
|
||||
self.version = '1.0'
|
||||
self.version = '1.1'
|
||||
|
||||
def is_encrypted(self, data):
|
||||
if data.startswith(HEADER):
|
||||
|
|
@ -59,7 +77,8 @@ class VaultLib(object):
|
|||
raise errors.AnsibleError("data is already encrypted")
|
||||
|
||||
if not self.cipher_name:
|
||||
raise errors.AnsibleError("the cipher must be set before encrypting data")
|
||||
self.cipher_name = "AES256"
|
||||
#raise errors.AnsibleError("the cipher must be set before encrypting data")
|
||||
|
||||
if 'Vault' + self.cipher_name in globals() and self.cipher_name in CIPHER_WHITELIST:
|
||||
cipher = globals()['Vault' + self.cipher_name]
|
||||
|
|
@ -67,13 +86,17 @@ class VaultLib(object):
|
|||
else:
|
||||
raise errors.AnsibleError("%s cipher could not be found" % self.cipher_name)
|
||||
|
||||
"""
|
||||
# combine sha + data
|
||||
this_sha = sha256(data).hexdigest()
|
||||
tmp_data = this_sha + "\n" + data
|
||||
"""
|
||||
|
||||
# encrypt sha + data
|
||||
tmp_data = this_cipher.encrypt(tmp_data, self.password)
|
||||
enc_data = this_cipher.encrypt(data, self.password)
|
||||
|
||||
# add header
|
||||
tmp_data = self._add_headers_and_hexify_encrypted_data(tmp_data)
|
||||
tmp_data = self._add_header(enc_data)
|
||||
return tmp_data
|
||||
|
||||
def decrypt(self, data):
|
||||
|
|
@ -83,8 +106,9 @@ class VaultLib(object):
|
|||
if not self.is_encrypted(data):
|
||||
raise errors.AnsibleError("data is not encrypted")
|
||||
|
||||
# clean out header, hex and sha
|
||||
data = self._split_headers_and_get_unhexified_data(data)
|
||||
# clean out header
|
||||
data = self._split_header(data)
|
||||
|
||||
|
||||
# create the cipher object
|
||||
if 'Vault' + self.cipher_name in globals() and self.cipher_name in CIPHER_WHITELIST:
|
||||
|
|
@ -96,33 +120,26 @@ class VaultLib(object):
|
|||
# try to unencrypt data
|
||||
data = this_cipher.decrypt(data, self.password)
|
||||
|
||||
# split out sha and verify decryption
|
||||
split_data = data.split("\n")
|
||||
this_sha = split_data[0]
|
||||
this_data = '\n'.join(split_data[1:])
|
||||
test_sha = sha256(this_data).hexdigest()
|
||||
if this_sha != test_sha:
|
||||
raise errors.AnsibleError("Decryption failed")
|
||||
return data
|
||||
|
||||
return this_data
|
||||
def _add_header(self, data):
|
||||
# combine header and encrypted data in 80 char columns
|
||||
|
||||
def _add_headers_and_hexify_encrypted_data(self, data):
|
||||
# combine header and hexlified encrypted data in 80 char columns
|
||||
|
||||
tmpdata = hexlify(data)
|
||||
tmpdata = [tmpdata[i:i+80] for i in range(0, len(tmpdata), 80)]
|
||||
#tmpdata = hexlify(data)
|
||||
tmpdata = [data[i:i+80] for i in range(0, len(data), 80)]
|
||||
|
||||
if not self.cipher_name:
|
||||
raise errors.AnsibleError("the cipher must be set before adding a header")
|
||||
|
||||
dirty_data = HEADER + ";" + str(self.version) + ";" + self.cipher_name + "\n"
|
||||
|
||||
for l in tmpdata:
|
||||
dirty_data += l + '\n'
|
||||
|
||||
return dirty_data
|
||||
|
||||
|
||||
def _split_headers_and_get_unhexified_data(self, data):
|
||||
def _split_header(self, data):
|
||||
# used by decrypt
|
||||
|
||||
tmpdata = data.split('\n')
|
||||
|
|
@ -130,14 +147,22 @@ class VaultLib(object):
|
|||
|
||||
self.version = str(tmpheader[1].strip())
|
||||
self.cipher_name = str(tmpheader[2].strip())
|
||||
clean_data = ''.join(tmpdata[1:])
|
||||
clean_data = '\n'.join(tmpdata[1:])
|
||||
|
||||
"""
|
||||
# strip out newline, join, unhex
|
||||
clean_data = [ x.strip() for x in clean_data ]
|
||||
clean_data = unhexlify(''.join(clean_data))
|
||||
"""
|
||||
|
||||
return clean_data
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *err):
|
||||
pass
|
||||
|
||||
class VaultEditor(object):
|
||||
# uses helper methods for write_file(self, filename, data)
|
||||
# to write a file so that code isn't duplicated for simple
|
||||
|
|
@ -153,6 +178,9 @@ class VaultEditor(object):
|
|||
def create_file(self):
|
||||
""" create a new encrypted file """
|
||||
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise errors.AnsibleError(CRYPTO_UPGRADE)
|
||||
|
||||
if os.path.isfile(self.filename):
|
||||
raise errors.AnsibleError("%s exists, please use 'edit' instead" % self.filename)
|
||||
|
||||
|
|
@ -166,6 +194,10 @@ class VaultEditor(object):
|
|||
self.write_data(enc_data, self.filename)
|
||||
|
||||
def decrypt_file(self):
|
||||
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise errors.AnsibleError(CRYPTO_UPGRADE)
|
||||
|
||||
if not os.path.isfile(self.filename):
|
||||
raise errors.AnsibleError("%s does not exist" % self.filename)
|
||||
|
||||
|
|
@ -179,6 +211,9 @@ class VaultEditor(object):
|
|||
|
||||
def edit_file(self):
|
||||
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise errors.AnsibleError(CRYPTO_UPGRADE)
|
||||
|
||||
# decrypt to tmpfile
|
||||
tmpdata = self.read_data(self.filename)
|
||||
this_vault = VaultLib(self.password)
|
||||
|
|
@ -191,9 +226,11 @@ class VaultEditor(object):
|
|||
call([EDITOR, tmp_path])
|
||||
new_data = self.read_data(tmp_path)
|
||||
|
||||
# create new vault and set cipher to old
|
||||
# create new vault
|
||||
new_vault = VaultLib(self.password)
|
||||
new_vault.cipher_name = this_vault.cipher_name
|
||||
|
||||
# we want the cipher to default to AES256
|
||||
#new_vault.cipher_name = this_vault.cipher_name
|
||||
|
||||
# encrypt new data a write out to tmp
|
||||
enc_data = new_vault.encrypt(new_data)
|
||||
|
|
@ -203,6 +240,10 @@ class VaultEditor(object):
|
|||
self.shuffle_files(tmp_path, self.filename)
|
||||
|
||||
def encrypt_file(self):
|
||||
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise errors.AnsibleError(CRYPTO_UPGRADE)
|
||||
|
||||
if not os.path.isfile(self.filename):
|
||||
raise errors.AnsibleError("%s does not exist" % self.filename)
|
||||
|
||||
|
|
@ -216,14 +257,20 @@ class VaultEditor(object):
|
|||
raise errors.AnsibleError("%s is already encrypted" % self.filename)
|
||||
|
||||
def rekey_file(self, new_password):
|
||||
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise errors.AnsibleError(CRYPTO_UPGRADE)
|
||||
|
||||
# decrypt
|
||||
tmpdata = self.read_data(self.filename)
|
||||
this_vault = VaultLib(self.password)
|
||||
dec_data = this_vault.decrypt(tmpdata)
|
||||
|
||||
# create new vault, set cipher to old and password to new
|
||||
# create new vault
|
||||
new_vault = VaultLib(new_password)
|
||||
new_vault.cipher_name = this_vault.cipher_name
|
||||
|
||||
# we want to force cipher to the default
|
||||
#new_vault.cipher_name = this_vault.cipher_name
|
||||
|
||||
# re-encrypt data and re-write file
|
||||
enc_data = new_vault.encrypt(dec_data)
|
||||
|
|
@ -254,11 +301,14 @@ class VaultEditor(object):
|
|||
|
||||
class VaultAES(object):
|
||||
|
||||
# this version has been obsoleted by the VaultAES256 class
|
||||
# which uses encrypt-then-mac (fixing order) and also improving the KDF used
|
||||
# code remains for upgrade purposes only
|
||||
# http://stackoverflow.com/a/16761459
|
||||
|
||||
def __init__(self):
|
||||
if not HAS_AES:
|
||||
raise errors.AnsibleError("pycrypto is not installed. Fix this with your package manager, for instance, yum-install python-crypto OR (apt equivalent)")
|
||||
raise errors.AnsibleError(CRYPTO_UPGRADE)
|
||||
|
||||
def aes_derive_key_and_iv(self, password, salt, key_length, iv_length):
|
||||
|
||||
|
|
@ -278,7 +328,12 @@ class VaultAES(object):
|
|||
|
||||
""" Read plaintext data from in_file and write encrypted to out_file """
|
||||
|
||||
in_file = BytesIO(data)
|
||||
|
||||
# combine sha + data
|
||||
this_sha = sha256(data).hexdigest()
|
||||
tmp_data = this_sha + "\n" + data
|
||||
|
||||
in_file = BytesIO(tmp_data)
|
||||
in_file.seek(0)
|
||||
out_file = BytesIO()
|
||||
|
||||
|
|
@ -301,14 +356,21 @@ class VaultAES(object):
|
|||
out_file.write(cipher.encrypt(chunk))
|
||||
|
||||
out_file.seek(0)
|
||||
return out_file.read()
|
||||
enc_data = out_file.read()
|
||||
tmp_data = hexlify(enc_data)
|
||||
|
||||
return tmp_data
|
||||
|
||||
|
||||
def decrypt(self, data, password, key_length=32):
|
||||
|
||||
""" Read encrypted data from in_file and write decrypted to out_file """
|
||||
|
||||
# http://stackoverflow.com/a/14989032
|
||||
|
||||
data = ''.join(data.split('\n'))
|
||||
data = unhexlify(data)
|
||||
|
||||
in_file = BytesIO(data)
|
||||
in_file.seek(0)
|
||||
out_file = BytesIO()
|
||||
|
|
@ -330,6 +392,129 @@ class VaultAES(object):
|
|||
|
||||
# reset the stream pointer to the beginning
|
||||
out_file.seek(0)
|
||||
return out_file.read()
|
||||
new_data = out_file.read()
|
||||
|
||||
# split out sha and verify decryption
|
||||
split_data = new_data.split("\n")
|
||||
this_sha = split_data[0]
|
||||
this_data = '\n'.join(split_data[1:])
|
||||
test_sha = sha256(this_data).hexdigest()
|
||||
|
||||
if this_sha != test_sha:
|
||||
raise errors.AnsibleError("Decryption failed")
|
||||
|
||||
#return out_file.read()
|
||||
return this_data
|
||||
|
||||
|
||||
class VaultAES256(object):
|
||||
|
||||
"""
|
||||
Vault implementation using AES-CTR with an HMAC-SHA256 authentication code.
|
||||
Keys are derived using PBKDF2
|
||||
"""
|
||||
|
||||
# http://www.daemonology.net/blog/2009-06-11-cryptographic-right-answers.html
|
||||
|
||||
def gen_key_initctr(self, password, salt):
|
||||
# 16 for AES 128, 32 for AES256
|
||||
keylength = 32
|
||||
|
||||
# match the size used for counter.new to avoid extra work
|
||||
ivlength = 16
|
||||
|
||||
hash_function = SHA256
|
||||
|
||||
# make two keys and one iv
|
||||
pbkdf2_prf = lambda p, s: HMAC.new(p, s, hash_function).digest()
|
||||
|
||||
if not HAS_PBKDF2:
|
||||
raise errors.AnsibleError(CRYPTO_UPGRADE)
|
||||
|
||||
derivedkey = PBKDF2(password, salt, dkLen=(2 * keylength) + ivlength,
|
||||
count=10000, prf=pbkdf2_prf)
|
||||
|
||||
#import epdb; epdb.st()
|
||||
key1 = derivedkey[:keylength]
|
||||
key2 = derivedkey[keylength:(keylength * 2)]
|
||||
iv = derivedkey[(keylength * 2):(keylength * 2) + ivlength]
|
||||
|
||||
return key1, key2, hexlify(iv)
|
||||
|
||||
|
||||
def encrypt(self, data, password):
|
||||
|
||||
salt = os.urandom(32)
|
||||
key1, key2, iv = self.gen_key_initctr(password, salt)
|
||||
|
||||
# PKCS#7 PAD DATA http://tools.ietf.org/html/rfc5652#section-6.3
|
||||
bs = AES.block_size
|
||||
padding_length = (bs - len(data) % bs) or bs
|
||||
data += padding_length * chr(padding_length)
|
||||
|
||||
# COUNTER.new PARAMETERS
|
||||
# 1) nbits (integer) - Length of the counter, in bits.
|
||||
# 2) initial_value (integer) - initial value of the counter. "iv" from gen_key_initctr
|
||||
|
||||
if not HAS_COUNTER:
|
||||
raise errors.AnsibleError(CRYPTO_UPGRADE)
|
||||
ctr = Counter.new(128, initial_value=long(iv, 16))
|
||||
|
||||
# AES.new PARAMETERS
|
||||
# 1) AES key, must be either 16, 24, or 32 bytes long -- "key" from gen_key_initctr
|
||||
# 2) MODE_CTR, is the recommended mode
|
||||
# 3) counter=<CounterObject>
|
||||
|
||||
cipher = AES.new(key1, AES.MODE_CTR, counter=ctr)
|
||||
|
||||
# ENCRYPT PADDED DATA
|
||||
cryptedData = cipher.encrypt(data)
|
||||
|
||||
# COMBINE SALT, DIGEST AND DATA
|
||||
hmac = HMAC.new(key2, cryptedData, SHA256)
|
||||
message = "%s\n%s\n%s" % ( hexlify(salt), hmac.hexdigest(), hexlify(cryptedData) )
|
||||
message = hexlify(message)
|
||||
return message
|
||||
|
||||
def decrypt(self, data, password):
|
||||
|
||||
# SPLIT SALT, DIGEST, AND DATA
|
||||
data = ''.join(data.split("\n"))
|
||||
data = unhexlify(data)
|
||||
salt, cryptedHmac, cryptedData = data.split("\n", 2)
|
||||
salt = unhexlify(salt)
|
||||
cryptedData = unhexlify(cryptedData)
|
||||
|
||||
key1, key2, iv = self.gen_key_initctr(password, salt)
|
||||
|
||||
# EXIT EARLY IF DIGEST DOESN'T MATCH
|
||||
hmacDecrypt = HMAC.new(key2, cryptedData, SHA256)
|
||||
if not self.is_equal(cryptedHmac, hmacDecrypt.hexdigest()):
|
||||
return None
|
||||
|
||||
# SET THE COUNTER AND THE CIPHER
|
||||
if not HAS_COUNTER:
|
||||
raise errors.AnsibleError(CRYPTO_UPGRADE)
|
||||
ctr = Counter.new(128, initial_value=long(iv, 16))
|
||||
cipher = AES.new(key1, AES.MODE_CTR, counter=ctr)
|
||||
|
||||
# DECRYPT PADDED DATA
|
||||
decryptedData = cipher.decrypt(cryptedData)
|
||||
|
||||
# UNPAD DATA
|
||||
padding_length = ord(decryptedData[-1])
|
||||
decryptedData = decryptedData[:-padding_length]
|
||||
|
||||
return decryptedData
|
||||
|
||||
def is_equal(self, a, b):
|
||||
# http://codahale.com/a-lesson-in-timing-attacks/
|
||||
if len(a) != len(b):
|
||||
return False
|
||||
|
||||
result = 0
|
||||
for x, y in zip(a, b):
|
||||
result |= ord(x) ^ ord(y)
|
||||
return result == 0
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -12,6 +12,21 @@ from nose.plugins.skip import SkipTest
|
|||
|
||||
from ansible import errors
|
||||
from ansible.utils.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
|
||||
|
|
@ -26,8 +41,8 @@ class TestVaultLib(TestCase):
|
|||
slots = ['is_encrypted',
|
||||
'encrypt',
|
||||
'decrypt',
|
||||
'_add_headers_and_hexify_encrypted_data',
|
||||
'_split_headers_and_get_unhexified_data',]
|
||||
'_add_header',
|
||||
'_split_header',]
|
||||
for slot in slots:
|
||||
assert hasattr(v, slot), "VaultLib is missing the %s method" % slot
|
||||
|
||||
|
|
@ -41,8 +56,7 @@ class TestVaultLib(TestCase):
|
|||
v = VaultLib('ansible')
|
||||
v.cipher_name = "TEST"
|
||||
sensitive_data = "ansible"
|
||||
sensitive_hex = hexlify(sensitive_data)
|
||||
data = v._add_headers_and_hexify_encrypted_data(sensitive_data)
|
||||
data = v._add_header(sensitive_data)
|
||||
lines = data.split('\n')
|
||||
assert len(lines) > 1, "failed to properly add header"
|
||||
header = lines[0]
|
||||
|
|
@ -52,19 +66,18 @@ class TestVaultLib(TestCase):
|
|||
assert header_parts[0] == '$ANSIBLE_VAULT', "header does not start with $ANSIBLE_VAULT"
|
||||
assert header_parts[1] == v.version, "header version is incorrect"
|
||||
assert header_parts[2] == 'TEST', "header does end with cipher name"
|
||||
assert lines[1] == sensitive_hex
|
||||
|
||||
def test_remove_header(self):
|
||||
def test_split_header(self):
|
||||
v = VaultLib('ansible')
|
||||
data = "$ANSIBLE_VAULT;9.9;TEST\n%s" % hexlify("ansible")
|
||||
rdata = v._split_headers_and_get_unhexified_data(data)
|
||||
data = "$ANSIBLE_VAULT;9.9;TEST\nansible"
|
||||
rdata = v._split_header(data)
|
||||
lines = rdata.split('\n')
|
||||
assert lines[0] == "ansible"
|
||||
assert v.cipher_name == 'TEST', "cipher name was not set"
|
||||
assert v.version == "9.9"
|
||||
|
||||
def test_encyrpt_decrypt(self):
|
||||
if not HAS_AES:
|
||||
def test_encrypt_decrypt_aes(self):
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise SkipTest
|
||||
v = VaultLib('ansible')
|
||||
v.cipher_name = 'AES'
|
||||
|
|
@ -73,8 +86,18 @@ class TestVaultLib(TestCase):
|
|||
assert enc_data != "foobar", "encryption failed"
|
||||
assert dec_data == "foobar", "decryption failed"
|
||||
|
||||
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'
|
||||
enc_data = v.encrypt("foobar")
|
||||
dec_data = v.decrypt(enc_data)
|
||||
assert enc_data != "foobar", "encryption failed"
|
||||
assert dec_data == "foobar", "decryption failed"
|
||||
|
||||
def test_encrypt_encrypted(self):
|
||||
if not HAS_AES:
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise SkipTest
|
||||
v = VaultLib('ansible')
|
||||
v.cipher_name = 'AES'
|
||||
|
|
@ -87,7 +110,7 @@ class TestVaultLib(TestCase):
|
|||
assert error_hit, "No error was thrown when trying to encrypt data with a header"
|
||||
|
||||
def test_decrypt_decrypted(self):
|
||||
if not HAS_AES:
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise SkipTest
|
||||
v = VaultLib('ansible')
|
||||
data = "ansible"
|
||||
|
|
@ -99,7 +122,8 @@ class TestVaultLib(TestCase):
|
|||
assert error_hit, "No error was thrown when trying to decrypt data without a header"
|
||||
|
||||
def test_cipher_not_set(self):
|
||||
if not HAS_AES:
|
||||
# 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"
|
||||
|
|
@ -108,6 +132,5 @@ class TestVaultLib(TestCase):
|
|||
enc_data = v.encrypt(data)
|
||||
except errors.AnsibleError, e:
|
||||
error_hit = True
|
||||
assert error_hit, "No error was thrown when trying to encrypt data without the cipher set"
|
||||
|
||||
|
||||
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
|
||||
|
|
|
|||
141
test/units/TestVaultEditor.py
Normal file
141
test/units/TestVaultEditor.py
Normal file
|
|
@ -0,0 +1,141 @@
|
|||
#!/usr/bin/env python
|
||||
|
||||
from unittest import TestCase
|
||||
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 import errors
|
||||
from ansible.utils.vault import VaultLib
|
||||
from ansible.utils.vault import VaultEditor
|
||||
|
||||
# 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 TestVaultEditor(TestCase):
|
||||
|
||||
def test_methods_exist(self):
|
||||
v = VaultEditor(None, None, None)
|
||||
slots = ['create_file',
|
||||
'decrypt_file',
|
||||
'edit_file',
|
||||
'encrypt_file',
|
||||
'rekey_file',
|
||||
'read_data',
|
||||
'write_data',
|
||||
'shuffle_files']
|
||||
for slot in slots:
|
||||
assert hasattr(v, slot), "VaultLib is missing the %s method" % slot
|
||||
|
||||
def test_decrypt_1_0(self):
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise SkipTest
|
||||
dirpath = tempfile.mkdtemp()
|
||||
filename = os.path.join(dirpath, "foo-ansible-1.0.yml")
|
||||
shutil.rmtree(dirpath)
|
||||
shutil.copytree("vault_test_data", dirpath)
|
||||
ve = VaultEditor(None, "ansible", filename)
|
||||
|
||||
# make sure the password functions for the cipher
|
||||
error_hit = False
|
||||
try:
|
||||
ve.decrypt_file()
|
||||
except errors.AnsibleError, e:
|
||||
error_hit = True
|
||||
|
||||
# verify decrypted content
|
||||
f = open(filename, "rb")
|
||||
fdata = f.read()
|
||||
f.close()
|
||||
|
||||
shutil.rmtree(dirpath)
|
||||
assert error_hit == False, "error decrypting 1.0 file"
|
||||
assert fdata.strip() == "foo", "incorrect decryption of 1.0 file: %s" % fdata.strip()
|
||||
|
||||
def test_decrypt_1_1(self):
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise SkipTest
|
||||
dirpath = tempfile.mkdtemp()
|
||||
filename = os.path.join(dirpath, "foo-ansible-1.1.yml")
|
||||
shutil.rmtree(dirpath)
|
||||
shutil.copytree("vault_test_data", dirpath)
|
||||
ve = VaultEditor(None, "ansible", filename)
|
||||
|
||||
# make sure the password functions for the cipher
|
||||
error_hit = False
|
||||
try:
|
||||
ve.decrypt_file()
|
||||
except errors.AnsibleError, e:
|
||||
error_hit = True
|
||||
|
||||
# verify decrypted content
|
||||
f = open(filename, "rb")
|
||||
fdata = f.read()
|
||||
f.close()
|
||||
|
||||
shutil.rmtree(dirpath)
|
||||
assert error_hit == False, "error decrypting 1.0 file"
|
||||
assert fdata.strip() == "foo", "incorrect decryption of 1.0 file: %s" % fdata.strip()
|
||||
|
||||
|
||||
def test_rekey_migration(self):
|
||||
if not HAS_AES or not HAS_COUNTER or not HAS_PBKDF2:
|
||||
raise SkipTest
|
||||
dirpath = tempfile.mkdtemp()
|
||||
filename = os.path.join(dirpath, "foo-ansible-1.0.yml")
|
||||
shutil.rmtree(dirpath)
|
||||
shutil.copytree("vault_test_data", dirpath)
|
||||
ve = VaultEditor(None, "ansible", filename)
|
||||
|
||||
# make sure the password functions for the cipher
|
||||
error_hit = False
|
||||
try:
|
||||
ve.rekey_file('ansible2')
|
||||
except errors.AnsibleError, e:
|
||||
error_hit = True
|
||||
|
||||
# verify decrypted content
|
||||
f = open(filename, "rb")
|
||||
fdata = f.read()
|
||||
f.close()
|
||||
|
||||
shutil.rmtree(dirpath)
|
||||
assert error_hit == False, "error rekeying 1.0 file to 1.1"
|
||||
|
||||
# ensure filedata can be decrypted, is 1.1 and is AES256
|
||||
vl = VaultLib("ansible2")
|
||||
dec_data = None
|
||||
error_hit = False
|
||||
try:
|
||||
dec_data = vl.decrypt(fdata)
|
||||
except errors.AnsibleError, e:
|
||||
error_hit = True
|
||||
|
||||
assert vl.cipher_name == "AES256", "wrong cipher name set after rekey: %s" % vl.cipher_name
|
||||
assert error_hit == False, "error decrypting migrated 1.0 file"
|
||||
assert dec_data.strip() == "foo", "incorrect decryption of rekeyed/migrated file: %s" % dec_data
|
||||
|
||||
|
||||
4
test/units/vault_test_data/foo-ansible-1.0.yml
Normal file
4
test/units/vault_test_data/foo-ansible-1.0.yml
Normal file
|
|
@ -0,0 +1,4 @@
|
|||
$ANSIBLE_VAULT;1.0;AES
|
||||
53616c7465645f5fd0026926a2d415a28a2622116273fbc90e377225c12a347e1daf4456d36a77f9
|
||||
9ad98d59f61d06a4b66718d855f16fb7bdfe54d1ec8aeaa4d06c2dc1fa630ae1846a029877f0eeb1
|
||||
83c62ffb04c2512995e815de4b4d29ed
|
||||
6
test/units/vault_test_data/foo-ansible-1.1.yml
Normal file
6
test/units/vault_test_data/foo-ansible-1.1.yml
Normal file
|
|
@ -0,0 +1,6 @@
|
|||
$ANSIBLE_VAULT;1.1;AES256
|
||||
62303130653266653331306264616235333735323636616539316433666463323964623162386137
|
||||
3961616263373033353631316333623566303532663065310a393036623466376263393961326530
|
||||
64336561613965383835646464623865663966323464653236343638373165343863623638316664
|
||||
3631633031323837340a396530313963373030343933616133393566366137363761373930663833
|
||||
3739
|
||||
Loading…
Reference in a new issue