Coverage for lib/ansible/parsing/vault/__init__.py : 17%

# (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 

# 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 os 

import random 

import shlex 

import shutil 

import subprocess 

import sys 

import tempfile 

import warnings 

from binascii import hexlify 

from binascii import unhexlify 

from binascii import Error as BinasciiError 

from hashlib import md5 

from hashlib import sha256 

from io import BytesIO 

HAS_CRYPTOGRAPHY = False 

HAS_PYCRYPTO = False 

HAS_SOME_PYCRYPTO = False 

CRYPTOGRAPHY_BACKEND = None 

try: 

with warnings.catch_warnings(): 

warnings.simplefilter("ignore", DeprecationWarning) 

from cryptography.exceptions import InvalidSignature 

from cryptography.hazmat.backends import default_backend 

from cryptography.hazmat.primitives import hashes, padding 

from cryptography.hazmat.primitives.hmac import HMAC 

from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC 

from cryptography.hazmat.primitives.ciphers import ( 

Cipher as C_Cipher, algorithms, modes 

) 

CRYPTOGRAPHY_BACKEND = default_backend() 

HAS_CRYPTOGRAPHY = True 

except ImportError: 

pass 

try: 

from Crypto.Cipher import AES as AES_pycrypto 

HAS_SOME_PYCRYPTO = True 

# Note: Only used for loading obsolete VaultAES files. All files are written 

# using the newer VaultAES256 which does not require md5 

from Crypto.Hash import SHA256 as SHA256_pycrypto 

from Crypto.Hash import HMAC as HMAC_pycrypto 

# Counter import fails for 2.0.1, requires >= 2.6.1 from pip 

from Crypto.Util import Counter as Counter_pycrypto 

# KDF import fails for 2.0.1, requires >= 2.6.1 from pip 

from Crypto.Protocol.KDF import PBKDF2 as PBKDF2_pycrypto 

HAS_PYCRYPTO = True 

except ImportError: 

pass 

from ansible.errors import AnsibleError, AnsibleAssertionError 

from ansible import constants as C 

from ansible.module_utils.six import PY3, binary_type 

# Note: on py2, this zip is izip not the list based zip() builtin 

from ansible.module_utils.six.moves import zip 

from ansible.module_utils._text import to_bytes, to_text, to_native 

try: 

from __main__ import display 

except ImportError: 

from ansible.utils.display import Display 

display = Display() 

b_HEADER = b'$ANSIBLE_VAULT' 

CIPHER_WHITELIST = frozenset((u'AES', u'AES256')) 

CIPHER_WRITE_WHITELIST = frozenset((u'AES256',)) 

# See also CIPHER_MAPPING at the bottom of the file which maps cipher strings 

# (used in VaultFile header) to a cipher class 

NEED_CRYPTO_LIBRARY = "ansible-vault requires either the cryptography library (preferred) or" 

96 ↛ 97line 96 didn't jump to line 97, because the condition on line 96 was never trueif HAS_SOME_PYCRYPTO: 

NEED_CRYPTO_LIBRARY += " a newer version of" 

NEED_CRYPTO_LIBRARY += " pycrypto in order to function." 

class AnsibleVaultError(AnsibleError): 

pass 

class AnsibleVaultPasswordError(AnsibleVaultError): 

pass 

class AnsibleVaultFormatError(AnsibleError): 

pass 

def is_encrypted(data): 

""" Test if this is vault encrypted data blob 

:arg data: a byte or text string to test whether it is recognized as vault 

encrypted data 

: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 

130 ↛ 131line 130 didn't jump to line 131, because the condition on line 130 was never true if b_data.startswith(b_HEADER): 

return True 

return False 

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. 

:arg file_obj: A file object that will be read from. 

: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 

current_position = file_obj.tell() 

try: 

file_obj.seek(start_pos) 

return is_encrypted(file_obj.read(count)) 

finally: 

file_obj.seek(current_position) 

def _parse_vaulttext_envelope(b_vaulttext_envelope, default_vault_id=None): 

b_tmpdata = b_vaulttext_envelope.splitlines() 

b_tmpheader = b_tmpdata[0].strip().split(b';') 

b_version = b_tmpheader[1].strip() 

cipher_name = to_text(b_tmpheader[2].strip()) 

vault_id = default_vault_id 

# Only attempt to find vault_id if the vault file is version 1.2 or newer 

# if self.b_version == b'1.2': 

if len(b_tmpheader) >= 4: 

vault_id = to_text(b_tmpheader[3].strip()) 

b_ciphertext = b''.join(b_tmpdata[1:]) 

return b_ciphertext, b_version, cipher_name, vault_id 

def parse_vaulttext_envelope(b_vaulttext_envelope, default_vault_id=None, filename=None): 

"""Parse the vaulttext envelope 

When data is saved, it has a header prepended and is formatted into 80 

character lines. This method extracts the information from the header 

and then removes the header and the inserted newlines. The string returned 

is suitable for processing by the Cipher classes. 

:arg b_vaulttext: byte str containing the data from a save file 

:kwarg default_vault_id: The vault_id name to use if the vaulttext does not provide one. 

:kwarg filename: The filename that the data came from. This is only 

used to make better error messages in case the data cannot be 

decrypted. This is optional. 

:returns: A tuple of byte str of the vaulttext suitable to pass to parse_vaultext, 

a byte str of the vault format version, 

the name of the cipher used, and the vault_id. 

:raises: AnsibleVaultFormatError: if the vaulttext_envelope format is invalid 

""" 

# used by decrypt 

default_vault_id = default_vault_id or C.DEFAULT_VAULT_IDENTITY 

try: 

return _parse_vaulttext_envelope(b_vaulttext_envelope, default_vault_id) 

except Exception as exc: 

msg = "Vault envelope format error" 

if filename: 

msg += ' in %s' % (filename) 

msg += ': %s' % exc 

raise AnsibleVaultFormatError(msg) 

def format_vaulttext_envelope(b_ciphertext, cipher_name, version=None, vault_id=None): 

""" Add header and format to 80 columns 

:arg b_ciphertext: the encrypted and hexlified data as a byte string 

:arg cipher_name: unicode cipher name (for ex, u'AES256') 

:arg version: unicode vault version (for ex, '1.2'). Optional ('1.1' is default) 

:arg vault_id: unicode vault identifier. If provided, the version will be bumped to 1.2. 

:returns: a byte str that should be dumped into a file. It's 

formatted to 80 char columns and has the header prepended 

""" 

if not cipher_name: 

raise AnsibleError("the cipher must be set before adding a header") 

version = version or '1.1' 

# If we specify a vault_id, use format version 1.2. For no vault_id, stick to 1.1 

if vault_id and vault_id != u'default': 

version = '1.2' 

b_version = to_bytes(version, 'utf-8', errors='strict') 

b_vault_id = to_bytes(vault_id, 'utf-8', errors='strict') 

b_cipher_name = to_bytes(cipher_name, 'utf-8', errors='strict') 

header_parts = [b_HEADER, 

b_version, 

b_cipher_name] 

if b_version == b'1.2' and b_vault_id: 

header_parts.append(b_vault_id) 

header = b';'.join(header_parts) 

b_vaulttext = [header] 

b_vaulttext += [b_ciphertext[i:i + 80] for i in range(0, len(b_ciphertext), 80)] 

b_vaulttext += [b''] 

b_vaulttext = b'\n'.join(b_vaulttext) 

return b_vaulttext 

def _unhexlify(b_data): 

try: 

return unhexlify(b_data) 

except (BinasciiError, TypeError) as exc: 

raise AnsibleVaultFormatError('Vault format unhexlify error: %s' % exc) 

def _parse_vaulttext(b_vaulttext): 

b_vaulttext = _unhexlify(b_vaulttext) 

b_salt, b_crypted_hmac, b_ciphertext = b_vaulttext.split(b"\n", 2) 

b_salt = _unhexlify(b_salt) 

b_ciphertext = _unhexlify(b_ciphertext) 

return b_ciphertext, b_salt, b_crypted_hmac 

def parse_vaulttext(b_vaulttext): 

"""Parse the vaulttext 

:arg b_vaulttext: byte str containing the vaulttext (ciphertext, salt, crypted_hmac) 

:returns: A tuple of byte str of the ciphertext suitable for passing to a 

Cipher class's decrypt() function, a byte str of the salt, 

and a byte str of the crypted_hmac 

:raises: AnsibleVaultFormatError: if the vaulttext format is invalid 

""" 

# SPLIT SALT, DIGEST, AND DATA 

try: 

return _parse_vaulttext(b_vaulttext) 

except AnsibleVaultFormatError: 

raise 

except Exception as exc: 

msg = "Vault vaulttext format error: %s" % exc 

raise AnsibleVaultFormatError(msg) 

def verify_secret_is_not_empty(secret, msg=None): 

'''Check the secret against minimal requirements. 

Raises: AnsibleVaultPasswordError if the password does not meet requirements. 

Currently, only requirement is that the password is not None or an empty string. 

''' 

msg = msg or 'Invalid vault password was provided' 

if not secret: 

raise AnsibleVaultPasswordError(msg) 

class VaultSecret: 

'''Opaque/abstract objects for a single vault secret. ie, a password or a key.''' 

def __init__(self, _bytes=None): 

# FIXME: ? that seems wrong... Unset etc? 

self._bytes = _bytes 

@property 

def bytes(self): 

'''The secret as a bytestring. 

Sub classes that store text types will need to override to encode the text to bytes. 

''' 

return self._bytes 

def load(self): 

return self._bytes 

class PromptVaultSecret(VaultSecret): 

default_prompt_formats = ["Vault password (%s): "] 

def __init__(self, _bytes=None, vault_id=None, prompt_formats=None): 

super(PromptVaultSecret, self).__init__(_bytes=_bytes) 

self.vault_id = vault_id 

if prompt_formats is None: 

self.prompt_formats = self.default_prompt_formats 

else: 

self.prompt_formats = prompt_formats 

@property 

def bytes(self): 

return self._bytes 

def load(self): 

self._bytes = self.ask_vault_passwords() 

def ask_vault_passwords(self): 

b_vault_passwords = [] 

for prompt_format in self.prompt_formats: 

prompt = prompt_format % {'vault_id': self.vault_id} 

try: 

vault_pass = display.prompt(prompt, private=True) 

except EOFError: 

raise AnsibleVaultError('EOFError (ctrl-d) on prompt for (%s)' % self.vault_id) 

verify_secret_is_not_empty(vault_pass) 

b_vault_pass = to_bytes(vault_pass, errors='strict', nonstring='simplerepr').strip() 

b_vault_passwords.append(b_vault_pass) 

# Make sure the passwords match by comparing them all to the first password 

for b_vault_password in b_vault_passwords: 

self.confirm(b_vault_passwords[0], b_vault_password) 

if b_vault_passwords: 

return b_vault_passwords[0] 

return None 

def confirm(self, b_vault_pass_1, b_vault_pass_2): 

# enforce no newline chars at the end of passwords 

if b_vault_pass_1 != b_vault_pass_2: 

# FIXME: more specific exception 

raise AnsibleError("Passwords do not match") 

def script_is_client(filename): 

'''Determine if a vault secret script is a client script that can be given --vault-id args''' 

# if password script is 'something-client' or 'something-client.[sh|py|rb|etc]' 

# script_name can still have '.' or could be entire filename if there is no ext 

script_name, dummy = os.path.splitext(filename) 

# TODO: for now, this is entirely based on filename 

if script_name.endswith('-client'): 

return True 

return False 

def get_file_vault_secret(filename=None, vault_id=None, encoding=None, loader=None): 

this_path = os.path.realpath(os.path.expanduser(filename)) 

if not os.path.exists(this_path): 

raise AnsibleError("The vault password file %s was not found" % this_path) 

if loader.is_executable(this_path): 

if script_is_client(filename): 

display.vvvv('The vault password file %s is a client script.' % filename) 

# TODO: pass vault_id_name to script via cli 

return ClientScriptVaultSecret(filename=this_path, vault_id=vault_id, 

encoding=encoding, loader=loader) 

# just a plain vault password script. No args, returns a byte array 

return ScriptVaultSecret(filename=this_path, encoding=encoding, loader=loader) 

return FileVaultSecret(filename=this_path, encoding=encoding, loader=loader) 

# TODO: mv these classes to a seperate file so we don't pollute vault with 'subprocess' etc 

class FileVaultSecret(VaultSecret): 

def __init__(self, filename=None, encoding=None, loader=None): 

super(FileVaultSecret, self).__init__() 

self.filename = filename 

self.loader = loader 

self.encoding = encoding or 'utf8' 

# We could load from file here, but that is eventually a pain to test 

self._bytes = None 

self._text = None 

@property 

def bytes(self): 

if self._bytes: 

return self._bytes 

if self._text: 

return self._text.encode(self.encoding) 

return None 

def load(self): 

self._bytes = self._read_file(self.filename) 

def _read_file(self, filename): 

""" 

Read a vault password from a file or if executable, execute the script and 

retrieve password from STDOUT 

""" 

# TODO: replace with use of self.loader 

try: 

f = open(filename, "rb") 

vault_pass = f.read().strip() 

f.close() 

except (OSError, IOError) as e: 

raise AnsibleError("Could not read vault password file %s: %s" % (filename, e)) 

b_vault_data, dummy = self.loader._decrypt_if_vault_data(vault_pass, filename) 

vault_pass = b_vault_data.strip(b'\r\n') 

verify_secret_is_not_empty(vault_pass, 

msg='Invalid vault password was provided from file (%s)' % filename) 

return vault_pass 

def __repr__(self): 

if self.filename: 

return "%s(filename='%s')" % (self.__class__.__name__, self.filename) 

return "%s()" % (self.__class__.__name__) 

class ScriptVaultSecret(FileVaultSecret): 

def _read_file(self, filename): 

if not self.loader.is_executable(filename): 

raise AnsibleVaultError("The vault password script %s was not executable" % filename) 

command = self._build_command() 

stdout, stderr, p = self._run(command) 

self._check_results(stdout, stderr, p) 

vault_pass = stdout.strip(b'\r\n') 

empty_password_msg = 'Invalid vault password was provided from script (%s)' % filename 

verify_secret_is_not_empty(vault_pass, 

msg=empty_password_msg) 

return vault_pass 

def _run(self, command): 

try: 

# STDERR not captured to make it easier for users to prompt for input in their scripts 

p = subprocess.Popen(command, stdout=subprocess.PIPE) 

except OSError as e: 

msg_format = "Problem running vault password script %s (%s)." \ 

" If this is not a script, remove the executable bit from the file." 

msg = msg_format % (self.filename, e) 

raise AnsibleError(msg) 

stdout, stderr = p.communicate() 

return stdout, stderr, p 

def _check_results(self, stdout, stderr, popen): 

if popen.returncode != 0: 

raise AnsibleError("Vault password script %s returned non-zero (%s): %s" % 

(self.filename, popen.returncode, stderr)) 

def _build_command(self): 

return [self.filename] 

class ClientScriptVaultSecret(ScriptVaultSecret): 

VAULT_ID_UNKNOWN_RC = 2 

def __init__(self, filename=None, encoding=None, loader=None, vault_id=None): 

super(ClientScriptVaultSecret, self).__init__(filename=filename, 

encoding=encoding, 

loader=loader) 

self._vault_id = vault_id 

display.vvvv('Executing vault password client script: %s --vault-id=%s' % (filename, vault_id)) 

def _run(self, command): 

try: 

p = subprocess.Popen(command, 

stdout=subprocess.PIPE, 

stderr=subprocess.PIPE) 

except OSError as e: 

msg_format = "Problem running vault password client script %s (%s)." \ 

" If this is not a script, remove the executable bit from the file." 

msg = msg_format % (self.filename, e) 

raise AnsibleError(msg) 

stdout, stderr = p.communicate() 

return stdout, stderr, p 

def _check_results(self, stdout, stderr, popen): 

if popen.returncode == self.VAULT_ID_UNKNOWN_RC: 

raise AnsibleError('Vault password client script %s did not find a secret for vault-id=%s: %s' % 

(self.filename, self._vault_id, stderr)) 

if popen.returncode != 0: 

raise AnsibleError("Vault password client script %s returned non-zero (%s) when getting secret for vault-id=%s: %s" % 

(self.filename, popen.returncode, self._vault_id, stderr)) 

def _build_command(self): 

command = [self.filename] 

if self._vault_id: 

command.extend(['--vault-id', self._vault_id]) 

return command 

def __repr__(self): 

if self.filename: 

return "%s(filename='%s', vault_id='%s')" % \ 

(self.__class__.__name__, self.filename, self._vault_id) 

return "%s()" % (self.__class__.__name__) 

def match_secrets(secrets, target_vault_ids): 

'''Find all VaultSecret objects that are mapped to any of the target_vault_ids in secrets''' 

if not secrets: 

return [] 

matches = [(vault_id, secret) for vault_id, secret in secrets if vault_id in target_vault_ids] 

return matches 

def match_best_secret(secrets, target_vault_ids): 

'''Find the best secret from secrets that matches target_vault_ids 

Since secrets should be ordered so the early secrets are 'better' than later ones, this 

just finds all the matches, then returns the first secret''' 

matches = match_secrets(secrets, target_vault_ids) 

if matches: 

return matches[0] 

# raise exception? 

return None 

def match_encrypt_vault_id_secret(secrets, encrypt_vault_id=None): 

# See if the --encrypt-vault-id matches a vault-id 

display.vvvv('encrypt_vault_id=%s' % encrypt_vault_id) 

if encrypt_vault_id is None: 

raise AnsibleError('match_encrypt_vault_id_secret requires a non None encrypt_vault_id') 

encrypt_vault_id_matchers = [encrypt_vault_id] 

encrypt_secret = match_best_secret(secrets, encrypt_vault_id_matchers) 

# return the best match for --encrypt-vault-id 

if encrypt_secret: 

return encrypt_secret 

# If we specified a encrypt_vault_id and we couldn't find it, dont 

# fallback to using the first/best secret 

raise AnsibleVaultError('Did not find a match for --encrypt-vault-id=%s in the known vault-ids %s' % (encrypt_vault_id, 

[_v for _v, _vs in secrets])) 

def match_encrypt_secret(secrets, encrypt_vault_id=None): 

'''Find the best/first/only secret in secrets to use for encrypting''' 

display.vvvv('encrypt_vault_id=%s' % encrypt_vault_id) 

# See if the --encrypt-vault-id matches a vault-id 

if encrypt_vault_id: 

return match_encrypt_vault_id_secret(secrets, 

encrypt_vault_id=encrypt_vault_id) 

# Find the best/first secret from secrets since we didnt specify otherwise 

# ie, consider all of the available secrets as matches 

_vault_id_matchers = [_vault_id for _vault_id, dummy in secrets] 

best_secret = match_best_secret(secrets, _vault_id_matchers) 

# can be empty list sans any tuple 

return best_secret 

class VaultLib: 

def __init__(self, secrets=None): 

self.secrets = secrets or [] 

self.cipher_name = None 

self.b_version = b'1.2' 

def encrypt(self, plaintext, secret=None, vault_id=None): 

"""Vault encrypt a piece of data. 

:arg plaintext: a text or byte string to encrypt. 

:returns: a utf-8 encoded byte str of encrypted data. The string 

contains a header identifying this as vault encrypted data and 

formatted to newline terminated lines of 80 characters. This is 

suitable for dumping as is to a vault file. 

If the string passed in is a text string, it will be encoded to UTF-8 

before encryption. 

""" 

if secret is None: 

if self.secrets: 

dummy, secret = match_encrypt_secret(self.secrets) 

else: 

raise AnsibleVaultError("A vault password must be specified to encrypt data") 

b_plaintext = to_bytes(plaintext, errors='surrogate_or_strict') 

if is_encrypted(b_plaintext): 

raise AnsibleError("input is already encrypted") 

if not self.cipher_name or self.cipher_name not in CIPHER_WRITE_WHITELIST: 

self.cipher_name = u"AES256" 

try: 

this_cipher = CIPHER_MAPPING[self.cipher_name]() 

except KeyError: 

raise AnsibleError(u"{0} cipher could not be found".format(self.cipher_name)) 

# encrypt data 

if vault_id: 

display.vvvvv('Encrypting with vault_id "%s" and vault secret %s' % (vault_id, secret)) 

else: 

display.vvvvv('Encrypting without a vault_id using vault secret %s' % secret) 

b_ciphertext = this_cipher.encrypt(b_plaintext, secret) 

# format the data for output to the file 

b_vaulttext = format_vaulttext_envelope(b_ciphertext, 

self.cipher_name, 

vault_id=vault_id) 

return b_vaulttext 

def decrypt(self, vaulttext, filename=None): 

'''Decrypt a piece of vault encrypted data. 

: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. 

: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 and the vault-id that was used 

''' 

plaintext, vault_id = self.decrypt_and_get_vault_id(vaulttext, filename=filename) 

return plaintext 

def decrypt_and_get_vault_id(self, vaulttext, filename=None): 

"""Decrypt a piece of vault encrypted data. 

: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. 

: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 and the vault-id that was used 

""" 

b_vaulttext = to_bytes(vaulttext, errors='strict', encoding='utf-8') 

if self.secrets is None: 

raise AnsibleVaultError("A vault password must be specified to decrypt data") 

if not is_encrypted(b_vaulttext): 

msg = "input is not vault encrypted data" 

if filename: 

msg += "%s is not a vault encrypted file" % to_native(filename) 

raise AnsibleError(msg) 

b_vaulttext, dummy, cipher_name, vault_id = parse_vaulttext_envelope(b_vaulttext, 

filename=filename) 

# create the cipher object, note that the cipher used for decrypt can 

# be different than the cipher used for encrypt 

if cipher_name in CIPHER_WHITELIST: 

this_cipher = CIPHER_MAPPING[cipher_name]() 

else: 

raise AnsibleError("{0} cipher could not be found".format(cipher_name)) 

b_plaintext = None 

if not self.secrets: 

raise AnsibleVaultError('Attempting to decrypt but no vault secrets found') 

# WARNING: Currently, the vault id is not required to match the vault id in the vault blob to 

# decrypt a vault properly. The vault id in the vault blob is not part of the encrypted 

# or signed vault payload. There is no cryptographic checking/verification/validation of the 

# vault blobs vault id. It can be tampered with and changed. The vault id is just a nick 

# name to use to pick the best secret and provide some ux/ui info. 

# iterate over all the applicable secrets (all of them by default) until one works... 

# if we specify a vault_id, only the corresponding vault secret is checked and 

# we check it first. 

vault_id_matchers = [] 

vault_id_used = None 

if vault_id: 

display.vvvvv('Found a vault_id (%s) in the vaulttext' % (vault_id)) 

vault_id_matchers.append(vault_id) 

_matches = match_secrets(self.secrets, vault_id_matchers) 

if _matches: 

display.vvvvv('We have a secret associated with vault id (%s), will try to use to decrypt %s' % (vault_id, to_text(filename))) 

else: 

display.vvvvv('Found a vault_id (%s) in the vault text, but we do not have a associated secret (--vault-id)' % (vault_id)) 

# Not adding the other secrets to vault_secret_ids enforces a match between the vault_id from the vault_text and 

# the known vault secrets. 

if not C.DEFAULT_VAULT_ID_MATCH: 

# Add all of the known vault_ids as candidates for decrypting a vault. 

vault_id_matchers.extend([_vault_id for _vault_id, _dummy in self.secrets if _vault_id != vault_id]) 

matched_secrets = match_secrets(self.secrets, vault_id_matchers) 

# for vault_secret_id in vault_secret_ids: 

for vault_secret_id, vault_secret in matched_secrets: 

display.vvvvv('Trying to use vault secret=(%s) id=%s to decrypt %s' % (vault_secret, vault_secret_id, to_text(filename))) 

try: 

# secret = self.secrets[vault_secret_id] 

display.vvvv('Trying secret %s for vault_id=%s' % (vault_secret, vault_secret_id)) 

b_plaintext = this_cipher.decrypt(b_vaulttext, vault_secret) 

if b_plaintext is not None: 

vault_id_used = vault_secret_id 

file_slug = '' 

if filename: 

file_slug = ' of "%s"' % filename 

display.vvvvv('Decrypt%s successful with secret=%s and vault_id=%s' % (file_slug, vault_secret, vault_secret_id)) 

break 

except AnsibleVaultFormatError as exc: 

msg = "There was a vault format error" 

if filename: 

msg += ' in %s' % (to_text(filename)) 

msg += ': %s' % exc 

display.warning(msg) 

raise 

except AnsibleError as e: 

display.vvvv('Tried to use the vault secret (%s) to decrypt (%s) but it failed. Error: %s' % 

(vault_secret_id, to_text(filename), e)) 

continue 

else: 

msg = "Decryption failed (no vault secrets were found that could decrypt)" 

if filename: 

msg += " on %s" % to_native(filename) 

raise AnsibleVaultError(msg) 

if b_plaintext is None: 

msg = "Decryption failed" 

if filename: 

msg += " on %s" % to_native(filename) 

raise AnsibleError(msg) 

return b_plaintext, vault_id_used 

class VaultEditor: 

def __init__(self, vault=None): 

# TODO: it may be more useful to just make VaultSecrets and index of VaultLib objects... 

self.vault = vault or VaultLib() 

# TODO: mv shred file stuff to it's own class 

def _shred_file_custom(self, tmp_path): 

""""Destroy a file, when shred (core-utils) is not available 

Unix `shred' destroys files "so that they can be recovered only with great difficulty with 

specialised hardware, if at all". It is based on the method from the paper 

"Secure Deletion of Data from Magnetic and Solid-State Memory", 

Proceedings of the Sixth USENIX Security Symposium (San Jose, California, July 22-25, 1996). 

We do not go to that length to re-implement shred in Python; instead, overwriting with a block 

of random data should suffice. 

See https://github.com/ansible/ansible/pull/13700 . 

""" 

file_len = os.path.getsize(tmp_path) 

if file_len > 0: # avoid work when file was empty 

max_chunk_len = min(1024 * 1024 * 2, file_len) 

passes = 3 

with open(tmp_path, "wb") as fh: 

for _ in range(passes): 

fh.seek(0, 0) 

# get a random chunk of data, each pass with other length 

chunk_len = random.randint(max_chunk_len // 2, max_chunk_len) 

data = os.urandom(chunk_len) 

for _ in range(0, file_len // chunk_len): 

fh.write(data) 

fh.write(data[:file_len % chunk_len]) 

# FIXME remove this assert once we have unittests to check its accuracy 

if fh.tell() != file_len: 

raise AnsibleAssertionError() 

os.fsync(fh) 

def _shred_file(self, tmp_path): 

"""Securely destroy a decrypted file 

Note standard limitations of GNU shred apply (For flash, overwriting would have no effect 

due to wear leveling; for other storage systems, the async kernel->filesystem->disk calls never 

guarantee data hits the disk; etc). Furthermore, if your tmp dirs is on tmpfs (ramdisks), 

it is a non-issue. 

Nevertheless, some form of overwriting the data (instead of just removing the fs index entry) is 

a good idea. If shred is not available (e.g. on windows, or no core-utils installed), fall back on 

a custom shredding method. 

""" 

if not os.path.isfile(tmp_path): 

# file is already gone 

return 

try: 

r = subprocess.call(['shred', tmp_path]) 

except (OSError, ValueError): 

# shred is not available on this system, or some other error occurred. 

# ValueError caught because OS X El Capitan is raising an 

# exception big enough to hit a limit in python2-2.7.11 and below. 

# Symptom is ValueError: insecure pickle when shred is not 

# installed there. 

r = 1 

if r != 0: 

# we could not successfully execute unix shred; therefore, do custom shred. 

self._shred_file_custom(tmp_path) 

os.remove(tmp_path) 

def _edit_file_helper(self, filename, secret, 

existing_data=None, force_save=False, vault_id=None): 

# Create a tempfile 

root, ext = os.path.splitext(os.path.realpath(filename)) 

fd, tmp_path = tempfile.mkstemp(suffix=ext) 

os.close(fd) 

try: 

if existing_data: 

self.write_data(existing_data, tmp_path, shred=False)