# Copyright 2018 AT&T Intellectual Property. All other rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import os from os import listdir from unittest import mock import pytest import yaml from pegleg import config from pegleg.engine.catalog.pki_generator import PKIGenerator from pegleg.engine.catalog import pki_utility from pegleg.engine import exceptions from pegleg.engine import secrets from pegleg.engine.util import encryption as crypt, git from pegleg.engine.util import files from pegleg.engine.util.pegleg_managed_document import \ PeglegManagedSecretsDocument from pegleg.engine.util.pegleg_secret_management import PeglegSecretManagement from tests.unit import test_utils from tests.unit.cli.test_commands import TEST_PARAMS TEST_DATA = """ --- schema: deckhand/Passphrase/v1 metadata: schema: metadata/Document/v1 name: osh_addons_keystone_ranger-agent_password layeringDefinition: abstract: false layer: site storagePolicy: encrypted data: 512363f37eab654313991174aef9f867d ... """ TEST_GLOBAL_DATA = """ --- schema: deckhand/Passphrase/v1 metadata: schema: metadata/Document/v1 name: osh_addons_keystone_ranger-agent_password layeringDefinition: abstract: false layer: global storagePolicy: encrypted data: 512363f37eab654313991174aef9f867d ... """ GLOBAL_PASSPHRASE_SALT_DOC = """ --- schema: deckhand/Passphrase/v1 metadata: schema: metadata/Document/v1 name: global_passphrase layeringDefinition: abstract: false layer: site storagePolicy: encrypted data: TbKYNtM@3gXpL=AFLAwU?&Ey ... --- schema: deckhand/Passphrase/v1 metadata: schema: metadata/Document/v1 name: global_salt layeringDefinition: abstract: false layer: site storagePolicy: encrypted data: h3=DQ#GNYEuCvybgpfW7ZxAP ... """ GLOBAL_SALT_DOC = """ --- schema: deckhand/Passphrase/v1 metadata: schema: metadata/Document/v1 name: global_salt layeringDefinition: abstract: false layer: site storagePolicy: encrypted data: h3=DQ#GNYEuCvybgpfW7ZxAP ... """ def test_encrypt_and_decrypt(): data = test_utils.rand_name( "this is an example of un-encrypted " "data.", "pegleg").encode() passphrase = test_utils.rand_name("passphrase1", "pegleg").encode() salt = test_utils.rand_name("salt1", "pegleg").encode() enc1 = crypt.encrypt(data, passphrase, salt) dec1 = crypt.decrypt(enc1, passphrase, salt) assert data == dec1 enc2 = crypt.encrypt(dec1, passphrase, salt) dec2 = crypt.decrypt(enc2, passphrase, salt) assert data == dec2 passphrase2 = test_utils.rand_name("passphrase2", "pegleg").encode() salt2 = test_utils.rand_name("salt2", "pegleg").encode() enc3 = crypt.encrypt(dec2, passphrase2, salt2) dec3 = crypt.decrypt(enc3, passphrase2, salt2) assert data == dec3 assert data != enc3 @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'aShortPassphrase', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_short_passphrase(): with pytest.raises(exceptions.PassphraseInsufficientLengthException): PeglegSecretManagement(file_path='file_path', author='test_author') @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'aShortSalt' }) def test_short_salt(): with pytest.raises(exceptions.SaltInsufficientLengthException): PeglegSecretManagement(file_path='file_path', author='test_author') @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_secret_encrypt_and_decrypt(temp_deployment_files, tmpdir): site_dir = tmpdir.join("deployment_files", "site", "cicd") passphrase_doc = """--- schema: deckhand/Passphrase/v1 metadata: schema: metadata/Document/v1 name: {0} storagePolicy: {1} layeringDefinition: abstract: False layer: {2} data: {0}-password ... """.format("cicd-passphrase-encrypted", "encrypted", "site") with open(os.path.join(str(site_dir), 'secrets', 'passphrases', 'cicd-passphrase-encrypted.yaml'), "w") \ as outfile: outfile.write(passphrase_doc) save_location = tmpdir.mkdir("encrypted_files") save_location_str = str(save_location) secrets.encrypt(save_location_str, "pytest", "cicd") encrypted_files = listdir(save_location_str) assert len(encrypted_files) > 0 encrypted_path = str( save_location.join( "site/cicd/secrets/passphrases/" "cicd-passphrase-encrypted.yaml")) decrypted = secrets.decrypt(encrypted_path) assert yaml.safe_load( decrypted[encrypted_path]) == yaml.safe_load(passphrase_doc) @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_pegleg_secret_management_constructor(): test_data = yaml.safe_load(TEST_DATA) doc = PeglegManagedSecretsDocument(test_data) assert doc.is_storage_policy_encrypted() assert not doc.is_encrypted() @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_pegleg_secret_management_constructor_with_invalid_arguments(): with pytest.raises(ValueError) as err_info: PeglegSecretManagement(file_path=None, docs=None) assert 'Either `file_path` or `docs` must be specified.' in str( err_info.value) with pytest.raises(ValueError) as err_info: PeglegSecretManagement(file_path='file_path', docs=['doc1']) assert 'Either `file_path` or `docs` must be specified.' in str( err_info.value) with pytest.raises(ValueError) as err_info: PeglegSecretManagement( file_path='file_path', generated=True, author='test_author') assert 'If the document is generated, author and catalog must be ' \ 'specified.' in str(err_info.value) with pytest.raises(ValueError) as err_info: PeglegSecretManagement(docs=['doc'], generated=True) assert 'If the document is generated, author and catalog must be ' \ 'specified.' in str(err_info.value) with pytest.raises(ValueError) as err_info: PeglegSecretManagement( docs=['doc'], generated=True, author='test_author') assert 'If the document is generated, author and catalog must be ' \ 'specified.' in str(err_info.value) with pytest.raises(ValueError) as err_info: PeglegSecretManagement(docs=['doc'], generated=True, catalog='catalog') assert 'If the document is generated, author and catalog must be ' \ 'specified.' in str(err_info.value) @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_pegleg_secret_management_double_encrypt(): encrypted_doc = PeglegSecretManagement( docs=[yaml.safe_load(TEST_DATA)]).get_encrypted_secrets()[0][0] encrypted_doc_2 = PeglegSecretManagement( docs=[encrypted_doc]).get_encrypted_secrets()[0][0] assert encrypted_doc == encrypted_doc_2 @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_encrypt_decrypt_using_file_path(tmpdir): # write the test data to temp file test_data = list(yaml.safe_load_all(TEST_DATA)) file_path = os.path.join(tmpdir, 'secrets_file.yaml') files.write(test_data, file_path) save_path = os.path.join(tmpdir, 'encrypted_secrets_file.yaml') # encrypt documents and validate that they were encrypted doc_mgr = PeglegSecretManagement(file_path=file_path, author='test_author') doc_mgr.encrypt_secrets(save_path) doc = doc_mgr.documents[0] assert doc.is_encrypted() assert doc.data['encrypted']['by'] == 'test_author' # decrypt documents and validate that they were decrypted doc_mgr = PeglegSecretManagement(file_path=file_path, author='test_author') doc_mgr.encrypt_secrets(save_path) # read back the encrypted file doc_mgr = PeglegSecretManagement(file_path=save_path, author='test_author') decrypted_data = doc_mgr.get_decrypted_secrets() assert test_data[0]['data'] == decrypted_data[0]['data'] assert test_data[0]['schema'] == decrypted_data[0]['schema'] @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_encrypt_decrypt_using_docs(tmpdir): # write the test data to temp file test_data = list(yaml.safe_load_all(TEST_DATA)) save_path = os.path.join(tmpdir, 'encrypted_secrets_file.yaml') # encrypt documents and validate that they were encrypted doc_mgr = PeglegSecretManagement(docs=test_data, author='test_author') doc_mgr.encrypt_secrets(save_path) doc = doc_mgr.documents[0] assert doc.is_encrypted() assert doc.data['encrypted']['by'] == 'test_author' # read back the encrypted file with open(save_path) as stream: encrypted_data = list(yaml.safe_load_all(stream)) # decrypt documents and validate that they were decrypted doc_mgr = PeglegSecretManagement(docs=encrypted_data, author='test_author') decrypted_data = doc_mgr.get_decrypted_secrets() assert test_data[0]['data'] == decrypted_data[0]['data'] assert test_data[0]['schema'] == decrypted_data[0]['schema'] assert test_data[0]['metadata']['name'] == decrypted_data[0]['metadata'][ 'name'] assert test_data[0]['metadata']['storagePolicy'] == decrypted_data[0][ 'metadata']['storagePolicy'] @pytest.mark.skipif( not pki_utility.PKIUtility.cfssl_exists(), reason='cfssl must be installed to execute these tests') @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_generate_pki_using_local_repo_path(temp_deployment_files): """Validates ``generate-pki`` action using local repo path.""" # Scenario: # # 1) Generate PKI using local repo path repo_path = str( git.git_handler(TEST_PARAMS["repo_url"], ref=TEST_PARAMS["repo_rev"])) with mock.patch.dict(config.GLOBAL_CONTEXT, {"site_repo": repo_path}): pki_generator = PKIGenerator( duration=365, sitename=TEST_PARAMS["site_name"]) generated_files = pki_generator.generate() assert len(generated_files), 'No secrets were generated' for generated_file in generated_files: with open(generated_file, 'r') as f: result = yaml.safe_load_all(f) # Validate valid YAML. assert list(result), "%s file is empty" % generated_file.name @pytest.mark.skipif( not pki_utility.PKIUtility.cfssl_exists(), reason='cfssl must be installed to execute these tests') @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_check_expiry(temp_deployment_files): """ Validates check_expiry """ repo_path = str( git.git_handler(TEST_PARAMS["repo_url"], ref=TEST_PARAMS["repo_rev"])) with mock.patch.dict(config.GLOBAL_CONTEXT, {"site_repo": repo_path}): pki_generator = PKIGenerator( duration=365, sitename=TEST_PARAMS["site_name"]) generated_files = pki_generator.generate() pki_util = pki_utility.PKIUtility(duration=0) assert len(generated_files), 'No secrets were generated' for generated_file in generated_files: if "certificate" not in generated_file: continue with open(generated_file, 'r') as f: results = yaml.safe_load_all(f) # Validate valid YAML. results = PeglegSecretManagement( docs=results).get_decrypted_secrets() for result in results: if result['schema'] == \ "deckhand/Certificate/v1": cert = result['data'] cert_info = pki_util.check_expiry(cert) assert cert_info['expired'] is False, \ "%s is expired/expiring on %s" % \ (result['metadata']['name'], cert_info['expiry_date']) @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_get_global_creds_missing_creds(temp_deployment_files, tmpdir): # Create site files site_dir = tmpdir.join("deployment_files", "site", "cicd") save_location = tmpdir.mkdir("encrypted_site_files") save_location_str = str(save_location) # Capture global credentials, verify they are not present and we default # to site credentials instead. config.set_global_enc_keys("cicd") passphrase, salt = secrets.get_global_creds("cicd") assert passphrase.decode() == 'ytrr89erARAiPE34692iwUMvWqqBvC' assert salt.decode() == 'MySecretSalt1234567890][' @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_get_global_creds_missing_pass(temp_deployment_files, tmpdir): # Create site files site_dir = tmpdir.join("deployment_files", "site", "cicd") # Create global salt file with open(os.path.join(str(site_dir), 'secrets', 'passphrases', 'cicd-global-salt-encrypted.yaml'), "w") as outfile: outfile.write(GLOBAL_SALT_DOC) save_location = tmpdir.mkdir("encrypted_site_files") save_location_str = str(save_location) # Demonstrate that encryption fails when only the global salt or # only the global passphrase are present among the site files. with pytest.raises(exceptions.GlobalCredentialsNotFound): config.set_global_enc_keys("cicd") @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_get_global_creds(temp_deployment_files, tmpdir): # Create site files site_dir = tmpdir.join("deployment_files", "site", "cicd") # Create global passphrase and salt file with open(os.path.join(str(site_dir), 'secrets', 'passphrases', 'cicd-global-passphrase-encrypted.yaml'), "w") \ as outfile: outfile.write(GLOBAL_PASSPHRASE_SALT_DOC) save_location = tmpdir.mkdir("encrypted_site_files") save_location_str = str(save_location) # Encrypt the global passphrase and salt file using site passphrase/salt config.set_global_enc_keys("cicd") secrets.encrypt(save_location_str, "pytest", "cicd") encrypted_files = listdir(save_location_str) assert len(encrypted_files) > 0 # Capture global credentials, verify we have the right ones passphrase, salt = secrets.get_global_creds("cicd") assert passphrase.decode() == "TbKYNtM@3gXpL=AFLAwU?&Ey" assert salt.decode() == "h3=DQ#GNYEuCvybgpfW7ZxAP" @mock.patch.dict( os.environ, { 'PEGLEG_PASSPHRASE': 'ytrr89erARAiPE34692iwUMvWqqBvC', 'PEGLEG_SALT': 'MySecretSalt1234567890][' }) def test_global_encrypt_decrypt(temp_deployment_files, tmpdir): # Create site files site_dir = tmpdir.join("deployment_files", "site", "cicd") # Create and encrypt global passphrase and salt file using site keys with open(os.path.join(str(site_dir), 'secrets', 'passphrases', 'cicd-global-passphrase-encrypted.yaml'), "w") \ as outfile: outfile.write(GLOBAL_PASSPHRASE_SALT_DOC) save_location = tmpdir.mkdir("encrypted_site_files") save_location_str = str(save_location) # Encrypt the global passphrase and salt file using site passphrase/salt config.set_global_enc_keys("cicd") secrets.encrypt(save_location_str, "pytest", "cicd") # Create and encrypt a global type document global_doc_path = os.path.join( str(site_dir), 'secrets', 'passphrases', 'globally_encrypted_doc.yaml') with open(global_doc_path, "w") as outfile: outfile.write(TEST_GLOBAL_DATA) # encrypt documents and validate that they were encrypted doc_mgr = PeglegSecretManagement( file_path=global_doc_path, author='pytest', site_name='cicd') doc_mgr.encrypt_secrets(global_doc_path) doc = doc_mgr.documents[0] assert doc.is_encrypted() assert doc.data['encrypted']['by'] == 'pytest' doc_mgr = PeglegSecretManagement( file_path=global_doc_path, author='pytest', site_name='cicd') decrypted_data = doc_mgr.get_decrypted_secrets() test_data = list(yaml.safe_load_all(TEST_GLOBAL_DATA)) assert test_data[0]['data'] == decrypted_data[0]['data'] assert test_data[0]['schema'] == decrypted_data[0]['schema']