airship/pegleg
Code Issues Proposed changes

Merge "Add pre-command decrypt option"

Browse Source
This commit is contained in:
Zuul 2019-12-20 20:16:07 +00:00 committed by Gerrit Code Review
commit 3bffa3bc07
5 changed files with 314 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
pegleg/cli/commands.py
View File

@ -98,9 +98,19 @@ def lint_repo(*, fail_on_missing_sub_src, exclude_lint, warn_lint):
@utils.EXTRA_REPOSITORY_OPTION @utils.EXTRA_REPOSITORY_OPTION
@utils.REPOSITORY_USERNAME_OPTION @utils.REPOSITORY_USERNAME_OPTION
@utils.REPOSITORY_KEY_OPTION @utils.REPOSITORY_KEY_OPTION
@click.option(
'--decrypt/--no-decrypt',
'decrypt_repos',
default=True,
help='Automatically attempts to decrypt repositories before executing '
'the command. Decryption will happen after repositories are copied to '
'the temporary directory created by pegleg or the user specified '
'`-p` directory. This means in most situations, pre-command decrypt '
'will not overwrite existing files. For overwriting existing files, '
'the full decrypt command should still be used.')
def site( def site(
*, site_repository, clone_path, extra_repositories, repo_key, *, site_repository, clone_path, extra_repositories, repo_key,
repo_username): repo_username, decrypt_repos):
"""Group for site-level actions, which include: """Group for site-level actions, which include:
* list: list available sites in a manifests repo * list: list available sites in a manifests repo
@ -115,7 +125,8 @@ def site(
repo_key, repo_key,
repo_username, repo_username,
extra_repositories or [], extra_repositories or [],
run_umask=True) run_umask=True,
decrypt_repos=decrypt_repos)
@site.command(help='Output complete config for one site') @site.command(help='Output complete config for one site')

10
pegleg/cli/utils.py
View File

@ -16,7 +16,9 @@ import logging
import click import click
from pegleg import config
from pegleg import engine from pegleg import engine
from pegleg import pegleg_main
LOG = logging.getLogger(__name__) LOG = logging.getLogger(__name__)
@ -40,9 +42,15 @@ def collection_default_callback(ctx, param, value):
return value return value
def decrypt_repos(site_name):
repo_list = config.all_repos()
for repo in repo_list:
pegleg_main.run_decrypt(True, repo, None, site_name)
# Arguments # # Arguments #
SITE_REPOSITORY_ARGUMENT = click.argument( SITE_REPOSITORY_ARGUMENT = click.argument(
'site_name', callback=process_repositories_callback) 'site_name', callback=process_repositories_callback, is_eager=True)
# Options # # Options #
ALLOW_MISSING_SUBSTITUTIONS_OPTION = click.option( ALLOW_MISSING_SUBSTITUTIONS_OPTION = click.option(

11
pegleg/config.py
View File

@ -38,7 +38,8 @@ except NameError:
'global_salt': None, 'global_salt': None,
'salt_min_length': 24, 'salt_min_length': 24,
'passphrase_min_length': 24, 'passphrase_min_length': 24,
'default_umask': 0o027 'default_umask': 0o027,
'decrypt_repos': False
} }
@ -214,3 +215,11 @@ def get_global_passphrase():
def get_global_salt(): def get_global_salt():
"""Get the global salt for encryption and decryption.""" """Get the global salt for encryption and decryption."""
return GLOBAL_CONTEXT['global_salt'] return GLOBAL_CONTEXT['global_salt']
def set_decrypt_repos(decrypt_repos=False):
GLOBAL_CONTEXT['decrypt_repos'] = decrypt_repos
def get_decrypt_repos():
return GLOBAL_CONTEXT['decrypt_repos']

32
pegleg/pegleg_main.py
View File

@ -49,7 +49,8 @@ def run_config(
repo_key, repo_key,
repo_username, repo_username,
extra_repositories, extra_repositories,
run_umask=True): run_umask=True,
decrypt_repos=True):
"""Initializes pegleg configuration data """Initializes pegleg configuration data
:param site_repository: path or URL for site repository :param site_repository: path or URL for site repository
@ -60,6 +61,7 @@ def run_config(
:param extra_repositories: list of extra repositories to read in documents :param extra_repositories: list of extra repositories to read in documents
from, specified as "type=REPO_URL/PATH" from, specified as "type=REPO_URL/PATH"
:param run_umask: if True, runs set_umask for os file output :param run_umask: if True, runs set_umask for os file output
:param decrypt_repos: if True, decrypts repos before executing command
:return: :return:
""" """
config.set_site_repo(site_repository) config.set_site_repo(site_repository)
@ -70,6 +72,7 @@ def run_config(
config.set_repo_username(repo_username) config.set_repo_username(repo_username)
if run_umask: if run_umask:
config.set_umask() config.set_umask()
config.set_decrypt_repos(decrypt_repos)
def _run_lint_helper( def _run_lint_helper(
@ -86,6 +89,20 @@ def _run_lint_helper(
return warns return warns
def _run_precommand_decrypt(site_name):
if config.get_decrypt_repos():
LOG.info('Executing pre-command repository decryption...')
repo_list = config.all_repos()
for repo in repo_list:
secrets_path = os.path.join(
repo.rstrip(os.path.sep), 'site', site_name, 'secrets')
if os.path.exists(secrets_path):
LOG.info('Decrypting %s', secrets_path)
run_decrypt(True, secrets_path, None, site_name)
else:
LOG.debug('Skipping pre-command repository decryption.')
def run_lint(exclude_lint, fail_on_missing_sub_src, warn_lint): def run_lint(exclude_lint, fail_on_missing_sub_src, warn_lint):
"""Runs linting on a repository """Runs linting on a repository
@ -116,6 +133,7 @@ def run_collect(exclude_lint, save_location, site_name, validate, warn_lint):
:param warn_lint: output warnings for specified rules :param warn_lint: output warnings for specified rules
:return: :return:
""" """
_run_precommand_decrypt(site_name)
if validate: if validate:
# Lint the primary repo prior to document collection. # Lint the primary repo prior to document collection.
_run_lint_helper( _run_lint_helper(
@ -154,6 +172,7 @@ def run_render(output_stream, site_name, validate):
:param validate: if True, validate documents using schema validation :param validate: if True, validate documents using schema validation
:return: :return:
""" """
_run_precommand_decrypt(site_name)
engine.site.render(site_name, output_stream, validate) engine.site.render(site_name, output_stream, validate)
@ -167,6 +186,7 @@ def run_lint_site(exclude_lint, fail_on_missing_sub_src, site_name, warn_lint):
:param warn_lint: output warnings for specified rules :param warn_lint: output warnings for specified rules
:return: :return:
""" """
_run_precommand_decrypt(site_name)
return _run_lint_helper( return _run_lint_helper(
fail_on_missing_sub_src=fail_on_missing_sub_src, fail_on_missing_sub_src=fail_on_missing_sub_src,
exclude_lint=exclude_lint, exclude_lint=exclude_lint,
@ -195,6 +215,7 @@ def run_upload(
:param site_name: site name to process :param site_name: site name to process
:return: response from shipyard instance :return: response from shipyard instance
""" """
_run_precommand_decrypt(site_name)
if not ctx.obj: if not ctx.obj:
ctx.obj = {} ctx.obj = {}
# Build API parameters required by Shipyard API Client. # Build API parameters required by Shipyard API Client.
@ -237,6 +258,7 @@ def run_generate_pki(
:param save_location: directory to store the generated site certificates in :param save_location: directory to store the generated site certificates in
:return: list of paths written to :return: list of paths written to
""" """
_run_precommand_decrypt(site_name)
engine.repository.process_repositories(site_name, overwrite_existing=True) engine.repository.process_repositories(site_name, overwrite_existing=True)
pkigenerator = catalog.pki_generator.PKIGenerator( pkigenerator = catalog.pki_generator.PKIGenerator(
site_name, site_name,
@ -264,7 +286,6 @@ def run_wrap_secret(
:param site_name: site name to process :param site_name: site name to process
:return: :return:
""" """
engine.repository.process_repositories(site_name, overwrite_existing=True)
config.set_global_enc_keys(site_name) config.set_global_enc_keys(site_name)
wrap_secret( wrap_secret(
author, author,
@ -285,6 +306,7 @@ def run_genesis_bundle(build_dir, site_name, validators):
:param validators: if True, runs validation scripts on genesis bundle :param validators: if True, runs validation scripts on genesis bundle
:return: :return:
""" """
_run_precommand_decrypt(site_name)
encryption_key = os.environ.get("PROMENADE_ENCRYPTION_KEY") encryption_key = os.environ.get("PROMENADE_ENCRYPTION_KEY")
config.set_global_enc_keys(site_name) config.set_global_enc_keys(site_name)
bundle.build_genesis( bundle.build_genesis(
@ -299,7 +321,7 @@ def run_check_pki_certs(days, site_name):
:param site_name: site name to process :param site_name: site name to process
:return: :return:
""" """
engine.repository.process_repositories(site_name, overwrite_existing=True) _run_precommand_decrypt(site_name)
config.set_global_enc_keys(site_name) config.set_global_enc_keys(site_name)
expiring_certs_exist, cert_results = engine.secrets.check_cert_expiry( expiring_certs_exist, cert_results = engine.secrets.check_cert_expiry(
site_name, duration=days) site_name, duration=days)
@ -335,7 +357,7 @@ def run_generate_passphrases(
discovered catalogs discovered catalogs
:return: :return:
""" """
engine.repository.process_repositories(site_name) _run_precommand_decrypt(site_name)
config.set_global_enc_keys(site_name) config.set_global_enc_keys(site_name)
engine.secrets.generate_passphrases( engine.secrets.generate_passphrases(
site_name, site_name,
@ -356,7 +378,6 @@ def run_encrypt(author, save_location, site_name):
:param site_name: site name to process :param site_name: site name to process
:return: :return:
""" """
engine.repository.process_repositories(site_name, overwrite_existing=True)
config.set_global_enc_keys(site_name) config.set_global_enc_keys(site_name)
if save_location is None: if save_location is None:
save_location = config.get_site_repo() save_location = config.get_site_repo()
@ -375,7 +396,6 @@ def run_decrypt(overwrite, path, save_location, site_name):
:rtype: list :rtype: list
""" """
decrypted_data = [] decrypted_data = []
engine.repository.process_repositories(site_name)
config.set_global_enc_keys(site_name) config.set_global_enc_keys(site_name)
decrypted = engine.secrets.decrypt(path, site_name=site_name) decrypted = engine.secrets.decrypt(path, site_name=site_name)
if overwrite: if overwrite:

280
tests/unit/cli/test_commands.py
View File

@ -11,14 +11,16 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import glob
import os import os
import subprocess
from unittest import mock from unittest import mock
from click.testing import CliRunner from click.testing import CliRunner
import pytest import pytest
import yaml import yaml
from pegleg import pegleg_main
from pegleg.cli import commands from pegleg.cli import commands
from pegleg.engine import errorcodes from pegleg.engine import errorcodes
from pegleg.engine.catalog import pki_utility from pegleg.engine.catalog import pki_utility
@ -97,7 +99,8 @@ class TestSiteCLIOptions(BaseCLIActionTest):
# Note that the -p option is used to specify the clone_folder # Note that the -p option is used to specify the clone_folder
site_list = self.runner.invoke( site_list = self.runner.invoke(
commands.site, ['-p', tmpdir, '-r', repo_url, 'list']) commands.site,
['--no-decrypt', '-p', tmpdir, '-r', repo_url, 'list'])
assert site_list.exit_code == 0 assert site_list.exit_code == 0
# Verify that the repo was cloned into the clone_path # Verify that the repo was cloned into the clone_path
@ -118,7 +121,8 @@ class TestSiteCLIOptions(BaseCLIActionTest):
# Note that the -p option is used to specify the clone_folder # Note that the -p option is used to specify the clone_folder
site_list = self.runner.invoke( site_list = self.runner.invoke(
commands.site, ['-p', tmpdir, '-r', repo_path, 'list']) commands.site,
['--no-decrypt', '-p', tmpdir, '-r', repo_path, 'list'])
assert site_list.exit_code == 0 assert site_list.exit_code == 0
# Verify that passing in clone_path when using local repo has no effect # Verify that passing in clone_path when using local repo has no effect
@ -146,14 +150,16 @@ class TestSiteCLIOptionsNegative(BaseCLIActionTest):
# Note that the -p option is used to specify the clone_folder # Note that the -p option is used to specify the clone_folder
site_list = self.runner.invoke( site_list = self.runner.invoke(
commands.site, ['-p', tmpdir, '-r', repo_url, 'list']) commands.site,
['--no-decrypt', '-p', tmpdir, '-r', repo_url, 'list'])
assert git.is_repository(os.path.join(tmpdir, self.repo_name)) assert git.is_repository(os.path.join(tmpdir, self.repo_name))
# Run site list for a second time to validate that the repo can't be # Run site list for a second time to validate that the repo can't be
# cloned twice in the same clone_path # cloned twice in the same clone_path
site_list = self.runner.invoke( site_list = self.runner.invoke(
commands.site, ['-p', tmpdir, '-r', repo_url, 'list']) commands.site,
['--no-decrypt', '-p', tmpdir, '-r', repo_url, 'list'])
assert site_list.exit_code == 1 assert site_list.exit_code == 1
assert 'File exists' in site_list.output assert 'File exists' in site_list.output
@ -167,8 +173,8 @@ class TestSiteCliActions(BaseCLIActionTest):
def _validate_collect_site_action(self, repo_path_or_url, save_location): def _validate_collect_site_action(self, repo_path_or_url, save_location):
result = self.runner.invoke( result = self.runner.invoke(
commands.site, [ commands.site, [
'-r', repo_path_or_url, 'collect', self.site_name, '-s', '--no-decrypt', '-r', repo_path_or_url, 'collect',
save_location self.site_name, '-s', save_location
]) ])
collected_files = os.listdir(save_location) collected_files = os.listdir(save_location)
@ -219,7 +225,9 @@ class TestSiteCliActions(BaseCLIActionTest):
def _test_lint_site_action(self, repo_path_or_url, exclude=True): def _test_lint_site_action(self, repo_path_or_url, exclude=True):
flag = '-x' if exclude else '-w' flag = '-x' if exclude else '-w'
lint_command = ['-r', repo_path_or_url, 'lint', self.site_name] lint_command = [
'--no-decrypt', '-r', repo_path_or_url, 'lint', self.site_name
]
exclude_lint_command = [ exclude_lint_command = [
flag, errorcodes.SCHEMA_STORAGE_POLICY_MISMATCH_FLAG, flag, flag, errorcodes.SCHEMA_STORAGE_POLICY_MISMATCH_FLAG, flag,
errorcodes.SECRET_NOT_ENCRYPTED_POLICY errorcodes.SECRET_NOT_ENCRYPTED_POLICY
@ -275,7 +283,10 @@ class TestSiteCliActions(BaseCLIActionTest):
def _validate_list_site_action(self, repo_path_or_url, tmpdir): def _validate_list_site_action(self, repo_path_or_url, tmpdir):
mock_output = os.path.join(tmpdir, 'output') mock_output = os.path.join(tmpdir, 'output')
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ['-r', repo_path_or_url, 'list', '-o', mock_output]) commands.site, [
'--no-decrypt', '-r', repo_path_or_url, 'list', '-o',
mock_output
])
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
with open(mock_output, 'r') as f: with open(mock_output, 'r') as f:
@ -309,8 +320,8 @@ class TestSiteCliActions(BaseCLIActionTest):
mock_output = os.path.join(tmpdir, 'output') mock_output = os.path.join(tmpdir, 'output')
result = self.runner.invoke( result = self.runner.invoke(
commands.site, [ commands.site, [
'-r', repo_path_or_url, 'show', self.site_name, '-o', '--no-decrypt', '-r', repo_path_or_url, 'show', self.site_name,
mock_output '-o', mock_output
]) ])
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
@ -340,7 +351,9 @@ class TestSiteCliActions(BaseCLIActionTest):
### Render tests ### ### Render tests ###
def _validate_render_site_action(self, repo_path_or_url): def _validate_render_site_action(self, repo_path_or_url):
render_command = ['-r', repo_path_or_url, 'render', self.site_name] render_command = [
'--no-decrypt', '-r', repo_path_or_url, 'render', self.site_name
]
with mock.patch('pegleg.engine.site.yaml') as mock_yaml: with mock.patch('pegleg.engine.site.yaml') as mock_yaml:
with mock.patch( with mock.patch(
@ -390,8 +403,8 @@ class TestSiteCliActions(BaseCLIActionTest):
with mock.patch('pegleg.pegleg_main.ShipyardHelper') as mock_obj: with mock.patch('pegleg.pegleg_main.ShipyardHelper') as mock_obj:
result = self.runner.invoke( result = self.runner.invoke(
commands.site, [ commands.site, [
'-r', repo_path, 'upload', self.site_name, '--collection', '--no-decrypt', '-r', repo_path, 'upload', self.site_name,
'collection' '--collection', 'collection'
]) ])
assert result.exit_code == 0 assert result.exit_code == 0
@ -413,7 +426,8 @@ class TestSiteCliActions(BaseCLIActionTest):
with mock.patch('pegleg.pegleg_main.ShipyardHelper') as mock_obj: with mock.patch('pegleg.pegleg_main.ShipyardHelper') as mock_obj:
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ['-r', repo_path, 'upload', self.site_name]) commands.site,
['--no-decrypt', '-r', repo_path, 'upload', self.site_name])
assert result.exit_code == 0 assert result.exit_code == 0
mock_obj.assert_called_once() mock_obj.assert_called_once()
@ -527,7 +541,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
secrets_opts = ['secrets', 'generate', 'certificates', self.site_name] secrets_opts = ['secrets', 'generate', 'certificates', self.site_name]
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ['-r', repo_url] + secrets_opts) commands.site, ['--no-decrypt', '-r', repo_url] + secrets_opts)
self._validate_generate_pki_action(result) self._validate_generate_pki_action(result)
@pytest.mark.skipif( @pytest.mark.skipif(
@ -543,7 +557,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
secrets_opts = ['secrets', 'generate', 'certificates', self.site_name] secrets_opts = ['secrets', 'generate', 'certificates', self.site_name]
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ['-r', repo_path] + secrets_opts) commands.site, ['--no-decrypt', '-r', repo_path] + secrets_opts)
self._validate_generate_pki_action(result) self._validate_generate_pki_action(result)
@pytest.mark.skipif( @pytest.mark.skipif(
@ -574,7 +588,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
secrets_opts = ['secrets', 'encrypt', '-a', 'test', self.site_name] secrets_opts = ['secrets', 'encrypt', '-a', 'test', self.site_name]
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ['-r', repo_path] + secrets_opts) commands.site, ['--no-decrypt', '-r', repo_path] + secrets_opts)
assert result.exit_code == 0 assert result.exit_code == 0
@ -590,7 +604,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
'secrets', 'decrypt', '--path', file_path, self.site_name 'secrets', 'decrypt', '--path', file_path, self.site_name
] ]
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ['-r', repo_path] + secrets_opts) commands.site, ['--no-decrypt', '-r', repo_path] + secrets_opts)
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
@pytest.mark.skipif( @pytest.mark.skipif(
@ -600,7 +614,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
repo_path = self.treasuremap_path repo_path = self.treasuremap_path
secrets_opts = ['secrets', 'check-pki-certs', self.site_name] secrets_opts = ['secrets', 'check-pki-certs', self.site_name]
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ['-r', repo_path] + secrets_opts) commands.site, ['--no-decrypt', '-r', repo_path] + secrets_opts)
assert result.exit_code == 1, result.output assert result.exit_code == 1, result.output
@pytest.mark.skipif( @pytest.mark.skipif(
@ -610,7 +624,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
repo_path = self.treasuremap_path repo_path = self.treasuremap_path
secrets_opts = ['secrets', 'check-pki-certs', 'airsloop'] secrets_opts = ['secrets', 'check-pki-certs', 'airsloop']
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ['-r', repo_path] + secrets_opts) commands.site, ['--no-decrypt', '-r', repo_path] + secrets_opts)
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
@mock.patch.dict( @mock.patch.dict(
@ -638,7 +652,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
"--no-encrypt", self.site_name "--no-encrypt", self.site_name
] ]
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ["-r", repo_path] + secrets_opts) commands.site, ['--no-decrypt', "-r", repo_path] + secrets_opts)
assert result.exit_code == 0 assert result.exit_code == 0
with open(output_path, "r") as output_fi: with open(output_path, "r") as output_fi:
@ -660,7 +674,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
"test-certificate", "-l", "site", self.site_name "test-certificate", "-l", "site", self.site_name
] ]
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ["-r", repo_path] + secrets_opts) commands.site, ['--no-decrypt', "-r", repo_path] + secrets_opts)
assert result.exit_code == 0 assert result.exit_code == 0
with open(output_path, "r") as output_fi: with open(output_path, "r") as output_fi:
@ -720,7 +734,10 @@ class TestSiteCliActionsWithSubdirectory(BaseCLIActionTest):
def _validate_list_site_action(self, repo_path_or_url, tmpdir): def _validate_list_site_action(self, repo_path_or_url, tmpdir):
mock_output = os.path.join(tmpdir, 'output') mock_output = os.path.join(tmpdir, 'output')
result = self.runner.invoke( result = self.runner.invoke(
commands.site, ['-r', repo_path_or_url, 'list', '-o', mock_output]) commands.site, [
'--no-decrypt', '-r', repo_path_or_url, 'list', '-o',
mock_output
])
with open(mock_output, 'r') as f: with open(mock_output, 'r') as f:
table_output = f.read() table_output = f.read()
@ -758,3 +775,218 @@ class TestSiteCliActionsWithSubdirectory(BaseCLIActionTest):
repo_path = os.path.join(_repo_path, 'deployment_files') repo_path = os.path.join(_repo_path, 'deployment_files')
self._validate_list_site_action(repo_path, tmpdir) self._validate_list_site_action(repo_path, tmpdir)
@pytest.mark.usefixtures('monkeypatch')
class TestCliSiteSubcommandsWithDecryptOption(BaseCLIActionTest):
@classmethod
def setup_class(cls):
super(TestCliSiteSubcommandsWithDecryptOption, cls).setup_class()
cls.runner = CliRunner(
env={
"PEGLEG_PASSPHRASE": 'ytrr89erARAiPE34692iwUMvWqqBvC',
"PEGLEG_SALT": "MySecretSalt1234567890][",
"PROMENADE_ENCRYPTION_KEY": "test"
})
for file in glob.iglob(os.path.join(cls.treasuremap_path, 'site',
'seaworthy', 'secrets', '**',
'*.yaml'), recursive=True):
args = [
'sed', '-i',
's/storagePolicy: cleartext/storagePolicy: encrypted/g', file
]
sed_output = subprocess.check_output(args, shell=False)
assert not sed_output
@mock.patch.dict(
os.environ, {
"PEGLEG_PASSPHRASE": 'ytrr89erARAiPE34692iwUMvWqqBvC',
"PEGLEG_SALT": "MySecretSalt1234567890]["
})
def setup(self):
pegleg_main.run_config(
self.treasuremap_path, None, None, None, [], True, False)
pegleg_main.run_encrypt('zuul-tester', None, self.site_name)
@staticmethod
def _validate_no_files_encrypted(path):
for file in glob.iglob(os.path.join(path, '**', '*.yaml'),
recursive=True):
with open(file, 'r') as f:
data = f.read()
if 'pegleg/PeglegManagedDocument/v1' in data:
return False
return True
def test_collect_using_decrypt_option(self, tmpdir):
"""Validates collect action using a path to a local repo."""
# Scenario:
#
# 1) Create temporary save location
# 2) Collect into save location (should skip clone repo)
# 3) Check that expected file name is there
repo_path = self.treasuremap_path
result = self.runner.invoke(
commands.site, [
'--decrypt', '-r', repo_path, 'collect', self.site_name, '-s',
tmpdir
])
collected_files = os.listdir(tmpdir)
assert result.exit_code == 0, result.output
assert len(collected_files) == 1
# Validates that site manifests collected from cloned repositories
# are written out to sensibly named files like airship-treasuremap.yaml
assert collected_files[0] == ("%s.yaml" % self.repo_name)
assert self._validate_no_files_encrypted(tmpdir)
def test_render_site_using_decrypt_option(self, tmpdir):
"""Validates render action using local repo path."""
# Scenario:
#
# 1) Mock out Deckhand render (so we can ignore P005 issues)
# 2) Render site (should skip clone repo)
repo_path = self.treasuremap_path
render_command = [
'--decrypt', '-p', tmpdir, '-r', repo_path, 'render',
self.site_name
]
with mock.patch('pegleg.engine.site.yaml') as mock_yaml:
with mock.patch(
'pegleg.engine.site.util.deckhand') as mock_deckhand:
mock_deckhand.deckhand_render.return_value = ([], [])
result = self.runner.invoke(commands.site, render_command)
assert result.exit_code == 0
mock_yaml.dump_all.assert_called_once()
assert self._validate_no_files_encrypted(
os.path.join(
tmpdir, 'treasuremap.git', 'site', 'seaworthy', 'secrets'))
def test_lint_site_using_decrypt_option(self, tmpdir):
"""Validates site lint action using local repo path."""
# Scenario:
#
# 1) Mock out Deckhand render (so we can ignore P005 issues)
# 2) Lint site with warn flags (should skip clone repo)
repo_path = self.treasuremap_path
lint_command = [
'--decrypt', '-p', tmpdir, '-r', repo_path, 'lint', self.site_name
]
exclude_lint_command = [
'-w', errorcodes.SCHEMA_STORAGE_POLICY_MISMATCH_FLAG, '-w',
errorcodes.SECRET_NOT_ENCRYPTED_POLICY
]
with mock.patch('pegleg.engine.site.util.deckhand') as mock_deckhand:
mock_deckhand.deckhand_render.return_value = ([], [])
result = self.runner.invoke(
commands.site, lint_command + exclude_lint_command)
assert result.exit_code == 0, result.output
assert self._validate_no_files_encrypted(
os.path.join(
tmpdir, 'treasuremap.git', 'site', 'seaworthy', 'secrets'))
@mock.patch.dict(
os.environ, {
"PEGLEG_PASSPHRASE": "123456789012345678901234567890",
"PEGLEG_SALT": "MySecretSalt1234567890]["
})
def test_upload_collection_callback_default_to_site_name(self, tmpdir):
"""Validates that collection will default to the given site_name"""
# Scenario:
#
# 1) Mock out ShipyardHelper
# 2) Check that ShipyardHelper was called with collection set to
# site_name
repo_path = self.treasuremap_path
with mock.patch('pegleg.pegleg_main.ShipyardHelper') as mock_obj:
result = self.runner.invoke(
commands.site, [
'--decrypt', '-p', tmpdir, '-r', repo_path, 'upload',
self.site_name
])
assert result.exit_code == 0
mock_obj.assert_called_once()
assert self._validate_no_files_encrypted(
os.path.join(
tmpdir, 'treasuremap.git', 'site', 'seaworthy', 'secrets'))
@pytest.mark.skipif(
not pki_utility.PKIUtility.cfssl_exists(),
reason='cfssl must be installed to execute these tests')
def test_site_secrets_generate_pki_using_decrypt_option(self, tmpdir):
"""Validates ``generate certificates`` action using local repo path."""
# Scenario:
#
# 1) Generate PKI using local repo path
repo_path = self.treasuremap_path
secrets_opts = ['secrets', 'generate', 'certificates', self.site_name]
result = self.runner.invoke(
commands.site,
['--decrypt', '-p', tmpdir, '-r', repo_path] + secrets_opts)
assert result.exit_code == 0
generated_files = []
output_lines = result.output.split("\n")
for line in output_lines:
if self.repo_name in line:
generated_files.append(line)
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
assert self._validate_no_files_encrypted(
os.path.join(
tmpdir, 'treasuremap.git', 'site', 'seaworthy', 'secrets'))
@pytest.mark.skipif(
not pki_utility.PKIUtility.cfssl_exists(),
reason='cfssl must be installed to execute these tests')
def test_check_pki_certs_expired_using_decrypt_option(self):
repo_path = self.treasuremap_path
secrets_opts = ['secrets', 'check-pki-certs', self.site_name]
result = self.runner.invoke(
commands.site, ['--decrypt', '-r', repo_path] + secrets_opts)
assert result.exit_code == 1, result.output
assert self._validate_no_files_encrypted(
os.path.join(repo_path, 'site', 'seaworthy', 'secrets'))
def test_genesis_bundle_using_decrypt_option(self, tmpdir):
repo_path = self.treasuremap_path
args = [
'--decrypt', '-p', tmpdir, '-r', repo_path, 'genesis_bundle', '-b',
tmpdir, self.site_name
]
with mock.patch(
'pegleg.pegleg_main.bundle.build_genesis') as mock_build:
result = self.runner.invoke(commands.site, args)
assert result.exit_code == 0
assert self._validate_no_files_encrypted(tmpdir)
mock_build.assert_called_once()
def test_generate_passphrases_using_decrypt_option(self, tmpdir):
repo_path = self.treasuremap_path
args = [
'--decrypt', '-p', tmpdir, '-r', repo_path, 'secrets', 'generate',
'passphrases', '-s', repo_path, '-a', 'zuul_tester', self.site_name
]
with mock.patch(
'pegleg.pegleg_main.engine.secrets.generate_passphrases'
) as mock_generator:
result = self.runner.invoke(commands.site, args)
assert result.exit_code == 0
assert self._validate_no_files_encrypted(tmpdir)
mock_generator.assert_called_once()