CLI capability to generate and encrypt passphrases

1. Adds the passphrases generation capability in Pegleg CLI,
so that pegleg can generation random passwords based on a
specification declared in pegleg/PassphrasesCatalog documents
2. Pegleg also wraps the generated passphrase documents in
pegleg managed documents, and encrypts the data.
3. Adds unit test cases for passphrase generation.
4. Updates pegleg CLI document.

Change-Id: I21d7668788cc24a8e0cc9cb0fb11df97600d0090
This commit is contained in:
pallav 2018-09-26 18:50:31 +05:30 committed by Lev Morgan
parent 1de8d5b68f
commit b79d5b7a98
25 changed files with 1186 additions and 111 deletions

View File

@ -613,6 +613,90 @@ Example:
/opt/security-manifests/site/site1/passwords/password1.yaml
generate
^^^^^^^^
A sub-group of secrets command group, which allows you to auto-generate
secrets documents of a site.
.. note::
The types of documents that pegleg cli generates are
passphrases, certificate authorities, certificates and keys. Passphrases are
declared in a new ``pegleg/PassphraseCatalog/v1`` document, while CAs,
certificates, and keys are declared in the ``pegleg/PKICatalog/v1``.
The ``pegleg/PKICatalog/v1`` schema is identical with the existing
``promenade/PKICatalog/v1``, promenade currently uses to generate the site
CAs, certificates, and keys.
The ``pegleg/PassphraseCatalog/v1`` schema is specified in
`Pegleg Passphrase Catalog`_
::
./pegleg.sh site -r <site_repo> -e <extra_repo> secrets generate <command> <options>
passphrases
"""""""""""
Generates, wraps and encrypts passphrase documents specified in the
``pegleg/PassphraseCatalog/v1`` document for a site. The site name, and the
directory to store the generated documents are provided by the
``site_name``, and the ``save_location`` command line parameters respectively.
The generated passphrases are stored in:
::
<save_location>/site/<site_name>/passphrases/<passphrase_name.yaml>
The schema for the generated passphrases is defined in
`Pegleg Managed Documents`_
**site_name** (Required).
Name of the ``site``. The ``site_name`` must match a ``site`` name in the site
repository folder structure. The ``generate`` command looks up the
``site-name``, and searches recursively the ``site_name`` folder structure
in the site repository for ``pegleg/PassphraseCatalog/v1`` documents. Then it
parses the passphrase catalog documents it found, and generates one passphrase
document for each passphrase ``document_name`` declared in the site passphrase
catalog.
**-a / --author** (Required)
``Author`` is intended to document the application or the individual, who
generates the site passphrase documents, mostly for tracking purposes. It
is expected to be leveraged in an operator-specific manner.
For instance the ``author`` can be the "userid" of the person running the
command, or the "application-id" of the application executing the command.
**-s / --save-location** (Required).
Where to output generated passphrase documents. The passphrase documents
are placed in the following folder structure under ``save_location``:
::
<save_location>/site/<site_name>/secrets/passphrases/<passphrase_name.yaml>
Usage:
::
./pegleg.sh site <options> secrets generate passphrases <site_name> -a
<author_id> -s <save_location>
Example
""""""""
::
./pegleg.sh site -r /opt/site-manifests \
-e global=/opt/manifests \
-e secrets=/opt/security-manifests \
secrets generate passphrases <site_name> -a <author_id> -s /workspace
CLI Repository Overrides
========================
@ -719,8 +803,9 @@ Where mandatory encrypted schema type is one of:
P002 - Deckhand rendering is expected to complete without errors.
P003 - All repos contain expected directories.
.. _Deckhand: https://airship-deckhand.readthedocs.io/en/latest/users/rendering.html
.. _Deckhand Validations: https://airship-deckhand.readthedocs.io/en/latest/overview.html#validation
.. _Deckhand: https://airship-deckhand.readthedocs.io/en/latest/rendering.html
.. _Deckhand Validations: https://airship-deckhand.readthedocs.io/en/latest/validation.html
.. _Pegleg Managed Documents: https://airship-specs.readthedocs.io/en/latest/specs/approved/pegleg-secrets.html#peglegmanageddocument
.. _Shipyard: https://github.com/openstack/airship-shipyard
.. _CLI documentation: https://airship-shipyard.readthedocs.io/en/latest/CLI.html#openstack-keystone-authorization-environment-variables
.. _Pegleg Passphrase Catalog: https://airship-specs.readthedocs.io/en/latest/specs/approved/pegleg-secrets.html#document-generation

View File

@ -71,3 +71,16 @@ PKI Exceptions
:members:
:show-inheritance:
:undoc-members:
Passphrase Exceptions
---------------------
.. autoexception:: pegleg.engine.exceptions.PassphraseSchemaNotFoundException
:members:
:show-inheritance:
:undoc-members:
.. autoexception:: pegleg.engine.exceptions.PassphraseCatalogNotFoundException
:members:
:show-inheritance:
:undoc-members:

View File

@ -57,17 +57,17 @@ EXTRA_REPOSITORY_OPTION = click.option(
'extra_repositories',
multiple=True,
help='Path or URL of additional repositories. These should be named per '
'the site-definition file, e.g. -e global=/opt/global -e '
'secrets=/opt/secrets. By default, the revision specified in the '
'site-definition for the site will be leveraged but can be overridden '
'using -e global=/opt/global@revision.')
'the site-definition file, e.g. -e global=/opt/global -e '
'secrets=/opt/secrets. By default, the revision specified in the '
'site-definition for the site will be leveraged but can be '
'overridden using -e global=/opt/global@revision.')
REPOSITORY_KEY_OPTION = click.option(
'-k',
'--repo-key',
'repo_key',
help='The SSH public key to use when cloning remote authenticated '
'repositories.')
'repositories.')
REPOSITORY_USERNAME_OPTION = click.option(
'-u',
@ -83,13 +83,15 @@ REPOSITORY_CLONE_PATH_OPTION = click.option(
'--clone-path',
'clone_path',
help='The path where the repo will be cloned. By default the repo will be '
'cloned to the /tmp path. If this option is included and the repo already '
'exists, then the repo will not be cloned again and the user must specify '
'a new clone path or pass in the local copy of the repository as the site '
'repository. Suppose the repo name is airship-treasuremap and the clone '
'path is /tmp/mypath then the following directory is created '
'/tmp/mypath/airship-treasuremap which will contain the contents of the '
'repo')
'cloned to the /tmp path. If this option is '
'included and the repo already '
'exists, then the repo will not be cloned again and the '
'user must specify a new clone path or pass in the local copy '
'of the repository as the site repository. Suppose the repo '
'name is airship-treasuremap and the clone path is '
'/tmp/mypath then the following directory is '
'created /tmp/mypath/airship-treasuremap '
'which will contain the contents of the repo')
ALLOW_MISSING_SUBSTITUTIONS_OPTION = click.option(
'-f',
@ -106,7 +108,7 @@ EXCLUDE_LINT_OPTION = click.option(
'exclude_lint',
multiple=True,
help='Excludes specified linting checks. Warnings will still be issued. '
'-w takes priority over -x.')
'-w takes priority over -x.')
WARN_LINT_OPTION = click.option(
'-w',
@ -225,7 +227,7 @@ def site(*, site_repository, clone_path, extra_repositories, repo_key,
'--save-location',
'save_location',
help='Directory to output the complete site definition. Created '
'automatically if it does not already exist.')
'automatically if it does not already exist.')
@click.option(
'--validate',
'validate',
@ -241,7 +243,7 @@ def site(*, site_repository, clone_path, extra_repositories, repo_key,
'exclude_lint',
multiple=True,
help='Excludes specified linting checks. Warnings will still be issued. '
'-w takes priority over -x.')
'-w takes priority over -x.')
@click.option(
'-w',
'--warn',
@ -344,8 +346,8 @@ def lint_site(*, fail_on_missing_sub_src, exclude_lint, warn_lint, site_name):
@click.option(
'--context-marker',
help='Specifies a UUID (8-4-4-4-12 format) that will be used to correlate '
'logs, transactions, etc. in downstream activities triggered by this '
'interaction ',
'logs, transactions, etc. in downstream activities triggered by this '
'interaction ',
required=False,
type=click.UUID)
@SITE_REPOSITORY_ARGUMENT
@ -375,24 +377,26 @@ def upload(ctx, *, os_project_domain_name,
click.echo(ShipyardHelper(ctx).upload_documents())
@site.group(name='secrets', help='Commands to manage site secrets documents')
@site.group(
name='secrets',
help='Commands to manage site secrets documents')
def secrets():
pass
@secrets.command(
'generate-pki',
help="""
Generate certificates and keys according to all PKICatalog documents in the
site. Regenerating certificates can be accomplished by re-running this command.
""")
help='Generate certificates and keys according to all PKICatalog '
'documents in the site. Regenerating certificates can be '
'accomplished by re-running this command.')
@click.option(
'-a',
'--author',
'author',
help="""Identifying name of the author generating new certificates. Used
for tracking provenance information in the PeglegManagedDocuments. An attempt
is made to automatically determine this value, but should be provided.""")
help='Identifying name of the author generating new certificates. Used'
'for tracking provenance information in the PeglegManagedDocuments. '
'An attempt is made to automatically determine this value, '
'but should be provided.')
@click.argument('site_name')
def generate_pki(site_name, author):
"""Generate certificates, certificate authorities and keypairs for a given
@ -442,27 +446,68 @@ def list_types(*, output_stream):
engine.type.list_types(output_stream)
@secrets.group(
name='generate',
help='Command group to generate site secrets documents.')
def generate():
pass
@generate.command(
'passphrases',
help='Command to generate site passphrases')
@click.argument('site_name')
@click.option(
'-s',
'--save-location',
'save_location',
required=True,
help='Directory to store the generated site passphrases in. It will '
'be created automatically, if it does not already exist. The '
'generated, wrapped, and encrypted passphrases files will be saved '
'in: <save_location>/site/<site_name>/secrets/passphrases/ '
'directory.')
@click.option(
'-a',
'--author',
'author',
required=True,
help='Identifier for the program or person who is generating the secrets '
'documents')
@click.option(
'-i',
'--interactive',
'interactive',
is_flag=bool,
default=False,
help='Generate passphrases interactively, not automatically')
def generate_passphrases(*, site_name, save_location, author, interactive):
engine.repository.process_repositories(site_name)
engine.secrets.generate_passphrases(site_name, save_location, author,
interactive)
@secrets.command(
'encrypt',
help='Command to encrypt and wrap site secrets '
'documents with metadata.storagePolicy set '
'to encrypted, in pegleg managed documents.')
'documents with metadata.storagePolicy set '
'to encrypted, in pegleg managed documents.')
@click.option(
'-s',
'--save-location',
'save_location',
default=None,
help='Directory to output the encrypted site secrets files. Created '
'automatically if it does not already exist. '
'If save_location is not provided, the output encrypted files will '
'overwrite the original input files (default behavior)')
'automatically if it does not already exist. '
'If save_location is not provided, the output encrypted files will '
'overwrite the original input files (default behavior)')
@click.option(
'-a',
'--author',
'author',
required=True,
help='Identifier for the program or person who is encrypting the secrets '
'documents')
'documents')
@click.argument('site_name')
def encrypt(*, save_location, author, site_name):
engine.repository.process_repositories(site_name, overwrite_existing=True)
@ -474,7 +519,7 @@ def encrypt(*, save_location, author, site_name):
@secrets.command(
'decrypt',
help='Command to unwrap and decrypt one site '
'secrets document and print it to stdout.')
'secrets document and print it to stdout.')
@click.option(
'-f',
'--filename',

View File

@ -26,7 +26,7 @@ except NameError:
'clone_path': None,
'site_path': 'site',
'site_rev': None,
'type_path': 'type',
'type_path': 'type'
}

View File

@ -24,8 +24,7 @@ from pegleg.engine.catalog import pki_utility
from pegleg.engine.common import managed_document as md
from pegleg.engine import exceptions
from pegleg.engine import util
from pegleg.engine.util.pegleg_managed_document import \
PeglegManagedSecretsDocument
from pegleg.engine.util.pegleg_secret_management import PeglegSecretManagement
__all__ = ['PKIGenerator']
@ -129,8 +128,8 @@ class PKIGenerator(object):
if not docs:
docs = generator(document_name, *args, **kwargs)
else:
docs = [PeglegManagedSecretsDocument(doc).pegleg_document
for doc in docs]
docs = PeglegSecretManagement(
docs=docs)
# Adding these to output should be idempotent, so we use a dict.
@ -215,6 +214,12 @@ class PKIGenerator(object):
LOG.debug('Creating secrets path: %s', dir_name)
os.makedirs(dir_name)
# Encrypt the document
document['data']['managedDocument']['metadata']['storagePolicy']\
= 'encrypted'
document = PeglegSecretManagement(docs=[
document]).get_encrypted_secrets()[0][0]
with open(output_path, 'a') as f:
# Don't use safe_dump so we can block format certificate
# data.

View File

@ -298,7 +298,8 @@ class PKIUtility(object):
'layeringDefinition': {
'abstract': False,
'layer': 'site',
}
},
'storagePolicy': 'cleartext'
}
wrapped_data = PKIUtility._block_literal(
data, block_strings=block_strings)

View File

View File

@ -0,0 +1,84 @@
# 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.
from abc import ABC
import logging
import os
import re
from pegleg import config
from pegleg.engine.exceptions import PassphraseCatalogNotFoundException
from pegleg.engine.util import definition
from pegleg.engine.util import git
LOG = logging.getLogger(__name__)
__all__ = ['BaseCatalog']
class BaseCatalog(ABC):
"""Abstract Base Class for all site catalogs."""
def __init__(self, kind, sitename, documents=None):
"""
Search for site catalog of the specified ``kind`` among the site
documents, and capture the catalog common metadata.
:param str kind: The catalog kind
:param str sitename: Name of the environment
:param list documents: Optional list of site documents. If not
present, the constructor will use the ``site_name` to lookup the list
of site documents.
"""
self._documents = documents or definition.documents_for_site(sitename)
self._site_name = sitename
self._catalog_path = None
self._kind = kind
self._catalog_docs = list()
for document in self._documents:
schema = document.get('schema')
if schema == 'pegleg/%s/v1' % kind:
self._catalog_docs.append(document)
elif schema == 'promenade/%s/v1' % kind:
LOG.warning('The schema promenade/%s/v1 is deprecated. Use '
'pegleg/%s/v1 instead.', kind, kind)
self._catalog_docs.append(document)
@property
def site_name(self):
return self._site_name
@property
def catalog_path(self):
if self._catalog_path is None:
self._set_catalog_path()
return self._catalog_path
def _set_catalog_path(self):
repo_name = git.repo_url(config.get_site_repo())
catalog_name = self._get_document_name('{}.yaml'.format(self._kind))
for file_path in definition.site_files(self.site_name):
if file_path.endswith(catalog_name) and repo_name in file_path:
self._catalog_path = os.path.join(
repo_name, file_path.split(repo_name)[1].lstrip('/'))
return
# Cound not find the Catalog for this generated passphrase
# raise an exception.
LOG.error('Catalog path: {} was not found in repo: {}'.format(
catalog_name, repo_name))
raise PassphraseCatalogNotFoundException()
def _get_document_name(self, name):
s1 = re.sub('(.)([A-Z][a-z]+)', r'\1-\2', name)
return re.sub('([a-z0-9])([A-Z])', r'\1-\2', s1).lower()

View File

@ -0,0 +1,88 @@
# 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 logging
from pegleg.engine.catalogs.base_catalog import BaseCatalog
from pegleg.engine.exceptions import PassphraseSchemaNotFoundException
LOG = logging.getLogger(__name__)
KIND = 'PassphraseCatalog'
P_DOCUMENT_NAME = 'document_name'
P_LENGTH = 'length'
P_DESCRIPTION = 'description'
P_ENCRYPTED = 'encrypted'
P_CLEARTEXT = 'cleartext'
P_DEFAULT_LENGTH = 24
P_DEFAULT_STORAGE_POLICY = 'encrypted'
__all__ = ['PassphraseCatalog']
class PassphraseCatalog(BaseCatalog):
"""Passphrase Catalog class.
The object containing methods and attributes to ingest and manage the site
passphrase catalog documents.
"""
def __init__(self, sitename, documents=None):
"""
Parse the site passphrase catalog documents and capture the
passphrase catalog data.
:param str sitename: Name of the environment
:param list documents: Environment configuration documents
:raises PassphraseSchemaNotFoundException: If it cannot find a
``pegleg/passphraseCatalog/v1`` document.
"""
super(PassphraseCatalog, self).__init__(KIND, sitename, documents)
if not self._catalog_docs:
raise PassphraseSchemaNotFoundException()
@property
def get_passphrase_names(self):
"""Return the list of passphrases in the catalog."""
return (passphrase[P_DOCUMENT_NAME]
for catalog in self._catalog_docs
for passphrase in catalog['data']['passphrases'])
def get_length(self, passphrase_name):
"""
Return the length of the ``passphrase_name``. If the catalog
does not specify a length for the ``passphrase_name``, return the
default passphrase length, 24.
"""
for c_doc in self._catalog_docs:
for passphrase in c_doc['data']['passphrases']:
if passphrase[P_DOCUMENT_NAME] == passphrase_name:
return passphrase.get(P_LENGTH, P_DEFAULT_LENGTH)
def get_storage_policy(self, passphrase_name):
"""
Return the storage policy of the ``passphrase_name``.
If the passphrase catalog does not specify a storage policy for
this passphrase, return the default storage policy, "encrypted".
"""
for c_doc in self._catalog_docs:
for passphrase in c_doc['data']['passphrases']:
if passphrase[P_DOCUMENT_NAME] == passphrase_name:
if P_ENCRYPTED in passphrase and not passphrase[
P_ENCRYPTED]:
return P_CLEARTEXT
else:
return P_DEFAULT_STORAGE_POLICY

View File

@ -75,3 +75,14 @@ class GitInvalidRepoException(PeglegBaseException):
class IncompletePKIPairError(PeglegBaseException):
"""Exception for incomplete private/public keypair."""
message = ("Incomplete keypair set %(kinds)s for name: %(name)s")
class PassphraseSchemaNotFoundException(PeglegBaseException):
"""Failed to find schema for Passphrases rendering."""
message = ('Could not find Passphrase schema for rendering Passphrases!')
class PassphraseCatalogNotFoundException(PeglegBaseException):
"""Failed to find Catalog for Passphrases generation."""
message = ('Could not find the Passphrase Catalog to generate '
'the site Passphrases!')

View File

View File

@ -0,0 +1,79 @@
# 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.
from abc import ABC
import logging
import os
from pegleg.engine import util
__all__ = ['BaseGenerator']
LOG = logging.getLogger(__name__)
class BaseGenerator(ABC):
"""
Abstract Base Class, providing the common data and methods for all
generator classes
"""
def __init__(self, sitename, save_location, author=None):
"""Constructor for ``BaseGenerator``.
:param str sitename: Name of the environment.
:param str save_location: The destination directory to store the
generated documents.
:param str author: Identifier for the individual or the application,
who requests to generate a document.
"""
self._sitename = sitename
self._documents = util.definition.documents_for_site(sitename)
self._save_location = save_location
self._author = author
@staticmethod
def generate_doc(kind, name, storage_policy, secret_data):
"""
Generate a document of the specified ``kind``, with the
specified ``storage_policy`` for the ``secret_data``.
:param str kind: Kind of the secret document.
:param str name: Name of the secret document
:param str storage_policy: Storage policy for the secret data
:param str secret_data: The data to be stored in this document.
"""
return {
'schema': 'deckhand/{}/v1'.format(kind),
'metadata': {
'schema': 'metadata/Document/v1',
'name': name,
'layeringDefinition': {
'abstract': False,
'layer': 'site',
},
'storagePolicy': storage_policy,
},
'data': secret_data,
}
def get_save_path(self, passphrase_name):
"""Calculate and return the save path of the ``passphrase_name``."""
return os.path.abspath(os.path.join(self._save_location,
'site',
self._sitename,
'secrets',
self.kind_path,
'{}.yaml'.format(passphrase_name)))

View File

@ -0,0 +1,90 @@
# 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.
from getpass import getpass
import logging
from pegleg.engine.catalogs import passphrase_catalog
from pegleg.engine.catalogs.passphrase_catalog import PassphraseCatalog
from pegleg.engine.generators.base_generator import BaseGenerator
from pegleg.engine.util import files
from pegleg.engine.util.passphrase import Passphrase
from pegleg.engine.util.pegleg_secret_management import PeglegSecretManagement
__all__ = ['PassphraseGenerator']
LOG = logging.getLogger(__name__)
KIND = 'Passphrase'
KIND_PATH = 'passphrases'
class PassphraseGenerator(BaseGenerator):
"""
Generates passphrases for a given environment, specified in a
passphrase catalog.
"""
def __init__(self, sitename, save_location, author):
"""Constructor for ``PassphraseGenerator``.
:param str sitename: Site name for which passphrases are generated.
:param str save_location: The base directory to store the generated
passphrase documents.
:param str author: Identifying name of the author generating new
certificates.
"""
super(PassphraseGenerator, self).__init__(
sitename, save_location, author)
self._catalog = PassphraseCatalog(
self._sitename, documents=self._documents)
self._pass_util = Passphrase()
def generate(self, interactive=False):
"""
For each passphrase entry in the passphrase catalog, generate a
random passphrase string, based on a passphrase specification in the
catalog. Create a pegleg managed document, wrap the generated
passphrase document in the pegleg managed document, and encrypt the
passphrase. Write the wrapped and encrypted document in a file at
<repo_name>/site/<site_name>/secrets/passphrases/passphrase_name.yaml.
"""
for p_name in self._catalog.get_passphrase_names:
passphrase = None
if interactive:
passphrase = getpass(
prompt="Input passphrase for {}. Leave blank to "
"auto-generate:\n".format(p_name))
if not passphrase:
passphrase = self._pass_util.get_pass(
self._catalog.get_length(p_name))
docs = list()
storage_policy = self._catalog.get_storage_policy(p_name)
docs.append(self.generate_doc(
KIND,
p_name,
storage_policy,
passphrase))
save_path = self.get_save_path(p_name)
if storage_policy == passphrase_catalog.P_ENCRYPTED:
PeglegSecretManagement(
docs=docs, generated=True, author=self._author,
catalog=self._catalog).encrypt_secrets(
save_path)
else:
files.write(save_path, docs)
@property
def kind_path(self):
return KIND_PATH

View File

@ -15,11 +15,12 @@
import logging
import os
from pegleg.engine.generators.passpharase_generator import PassphraseGenerator
from pegleg.engine.util import definition
from pegleg.engine.util import files
from pegleg.engine.util.pegleg_secret_management import PeglegSecretManagement
__all__ = ('encrypt', 'decrypt')
__all__ = ('encrypt', 'decrypt', 'generate_passphrases')
LOG = logging.getLogger(__name__)
@ -28,22 +29,21 @@ def encrypt(save_location, author, site_name):
"""
Encrypt all secrets documents for a site identifies by site_name.
Parse through all documents related to site_name and encrypt all
site documents which have metadata.storagePolicy: encrypted, and which are
not already encrypted and wrapped in a PeglegManagedDocument.
Passphrase and salt for the encryption are read from environment
variables ($PEGLEG_PASSPHRASE and $PEGLEG_SALT respectively).
Parse through all documents related to ``site_name`` and encrypt all
site documents, which have metadata.storagePolicy: encrypted, and
are not already encrypted and wrapped in a PeglegManagedDocument.
``Passphrase`` and ``salt`` for the encryption are read from environment
variables``$PEGLEG_PASSPHRASE`` and ``$PEGLEG_SALT`` respectively.
By default, the resulting output files will overwrite the original
unencrypted secrets documents.
:param save_location: if provided, identifies the base directory to store
the encrypted secrets files. If not provided the encrypted secrets files
will overwrite the original unencrypted files (default behavior).
:type save_location: string
:param author: The identifier provided by the application or
the person who requests encrypt the site secrets documents.
:type author: string
:param site_name: The name of the site to encrypt its secrets files.
:type site_name: string
:param str save_location: if provided, is used as the base directory to
store the encrypted secrets files. If not provided, the encrypted
secrets files will overwrite the original unencrypted files (default
behavior).
:param str author: Identifies the individual or application, who
encrypts the secrets documents.
:param str site_name: The name of the site to encrypt its secrets files.
"""
files.check_file_save_location(save_location)
@ -51,8 +51,9 @@ def encrypt(save_location, author, site_name):
secrets_found = False
for repo_base, file_path in definition.site_files_by_repo(site_name):
secrets_found = True
PeglegSecretManagement(file_path).encrypt_secrets(
_get_dest_path(repo_base, file_path, save_location), author)
PeglegSecretManagement(
file_path=file_path, author=author).encrypt_secrets(
_get_dest_path(repo_base, file_path, save_location))
if secrets_found:
LOG.info('Encryption of all secret files was completed.')
else:
@ -62,11 +63,11 @@ def encrypt(save_location, author, site_name):
def decrypt(file_path, site_name):
"""
Decrypt one secrets file and print the decrypted data to standard out.
Decrypt one secrets file, and print the decrypted file to standard out.
Search in in secrets file of a site, identified by site_name, for a file
named file_name.
If the file is found and encrypted, unwrap and decrypt it and print the
Search in secrets file of a site, identified by ``site_name``, for a file
named ``file_name``.
If the file is found and encrypted, unwrap and decrypt it, and print the
result to standard out.
If the file is found, but it is not encrypted, print the contents of the
file to standard out.
@ -90,7 +91,7 @@ def decrypt(file_path, site_name):
def _get_dest_path(repo_base, file_path, save_location):
"""
Calculate and return the destination base directory path for the
encrypted or decrypted secrets files.
encrypted secrets files.
:param repo_base: Base repo of the source secrets file.
:type repo_base: string
@ -111,3 +112,20 @@ def _get_dest_path(repo_base, file_path, save_location):
return file_path.replace(repo_base, save_location)
else:
return file_path
def generate_passphrases(site_name, save_location, author, interactive=False):
"""
Look for the site passphrase catalogs, and for every passphrase entry in
the passphrase catalog generate a passphrase document, wrap the
passphrase document in a pegleg managed document, and encrypt the
passphrase data.
:param interactive: Whether to generate the results interactively
:param str site_name: The site to read from
:param str save_location: Location to write files to
:param str author:
"""
PassphraseGenerator(site_name, save_location, author).generate(
interactive=interactive)

View File

@ -25,6 +25,8 @@ KEY_LENGTH = 32
ITERATIONS = 10000
LOG = logging.getLogger(__name__)
__all__ = ('encrypt', 'decrypt')
def encrypt(unencrypted_data,
passphrase,

View File

@ -141,6 +141,14 @@ def _get_current_ref(repo_url):
return None
def get_remote_url(repo_url):
try:
repo = Repo(repo_url, search_parent_directories=True)
return repo.remotes.origin.url
except Exception as e:
return None
def _try_git_clone(repo_url,
ref=None,
proxy_server=None,

View File

@ -0,0 +1,33 @@
# 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.
from random import SystemRandom
from rstr import Rstr
import string
__all__ = ['Passphrase']
class Passphrase(object):
def __init__(self):
self._pool = string.ascii_letters + string.digits + string.punctuation
self._rs = Rstr(SystemRandom())
def get_pass(self, pass_len=24):
"""Create and return a random password, of the ``pass_len`` length."""
if pass_len < 24:
pass_len = 24
return self._rs.rstr(self._pool, pass_len)

View File

@ -15,48 +15,66 @@
from datetime import datetime
import logging
from pegleg import config
from pegleg.engine.util import git
PEGLEG_MANAGED_SCHEMA = 'pegleg/PeglegManagedDocument/v1'
ENCRYPTED = 'encrypted'
GENERATED = 'generated'
STORAGE_POLICY = 'storagePolicy'
METADATA = 'metadata'
LOG = logging.getLogger(__name__)
__all__ = ['PeglegManagedSecretsDocument']
class PeglegManagedSecretsDocument(object):
"""Object representing one Pegleg managed secret document."""
def __init__(self, secrets_document):
def __init__(self, document, generated=False, catalog=None, author=None):
"""
Parse and wrap an externally generated document in a
pegleg managed document.
:param secrets_document: The content of the source document
:type secrets_document: dict
:param document: The content of the source document
:type document: dict
:param bool generated: A flag to indicate the documents are
auto-generated by pegleg (True), or manually created (False).
:param catalog: catalog of the generated secret documents. A catalog
must be provided, only if generated is True.
:type catalog: A subclass of the ABC
pegleg.catalogs.base_catalog.BaseCatalog
"""
if self.is_pegleg_managed_secret(secrets_document):
self._pegleg_document = secrets_document
self._catalog = catalog
self._author = author
self._generated = generated
if self.is_pegleg_managed_secret(document):
self._pegleg_document = document
else:
self._pegleg_document =\
self.__wrap(secrets_document)
self._pegleg_document = self.__wrap(
document, generated, catalog, author)
self._embedded_document = \
self._pegleg_document['data']['managedDocument']
@staticmethod
def __wrap(secrets_document):
def __wrap(secrets_document, generated=False, catalog=None, author=None):
"""
Embeds a valid deckhand document in a pegleg managed document.
:param secrets_document: secrets document to be embedded in a
pegleg managed document.
:type secrets_document: dict
:param bool generated: A flag to indicate the documents are
auto-generated by pegleg (True), or manually created (False).
:return: pegleg manged document with the wrapped original secrets
document.
:rtype: dict
"""
return {
doc = {
'schema': PEGLEG_MANAGED_SCHEMA,
'metadata': {
'name': secrets_document['metadata']['name'],
@ -78,6 +96,18 @@ class PeglegManagedSecretsDocument(object):
}
}
if generated:
doc['data'][GENERATED] = {
'at': datetime.utcnow().isoformat(),
'by': author,
'specifiedBy': {
'repo': git.repo_url(config.get_site_repo()),
'reference': config.get_site_rev() or 'master',
'path': catalog.catalog_path,
},
}
return doc
@staticmethod
def is_pegleg_managed_secret(secrets_document):
""""
@ -117,18 +147,24 @@ class PeglegManagedSecretsDocument(object):
otherwise."""
return ENCRYPTED in self.data
def is_generated(self):
"""If the document is already marked auto-generated return True. False
otherwise."""
return GENERATED in self.data
def is_storage_policy_encrypted(self):
"""If the document's storagePolicy is set to encrypted return True.
False otherwise."""
return STORAGE_POLICY in self._embedded_document[METADATA] \
and ENCRYPTED in self._embedded_document[METADATA][STORAGE_POLICY]
def set_encrypted(self, author):
def set_encrypted(self, author=None):
"""Mark the pegleg managed document as encrypted."""
self.data[ENCRYPTED] = {
'at': datetime.utcnow().isoformat(),
'by': author,
'at': datetime.utcnow().isoformat()
}
if author:
self.data[ENCRYPTED]['by'] = author
def set_decrypted(self):
"""Mark the pegleg managed document as un-encrypted."""

View File

@ -34,7 +34,8 @@ ENV_SALT = 'PEGLEG_SALT'
class PeglegSecretManagement(object):
"""An object to handle operations on of a pegleg managed file."""
def __init__(self, file_path=None, docs=None):
def __init__(self, file_path=None, docs=None, generated=False,
catalog=None, author=None):
"""
Read the source file and the environment data needed to wrap and
process the file documents as pegleg managed document.
@ -43,22 +44,40 @@ class PeglegSecretManagement(object):
"""
if all([file_path, docs]) or not any([file_path, docs]):
raise ValueError('Either `file_path` or `docs` must be specified.')
raise ValueError('Either `file_path` or `docs` must be '
'specified.')
if generated and not (author and catalog):
raise ValueError("If the document is generated, author and "
"catalog must be specified.")
self.__check_environment()
self.file_path = file_path
self.documents = list()
self._generated = generated
if docs:
for doc in docs:
self.documents.append(PeglegManagedSecret(doc))
self.documents.append(PeglegManagedSecret(doc,
generated=generated,
catalog=catalog,
author=author))
else:
self.file_path = file_path
for doc in files.read(file_path):
self.documents.append(PeglegManagedSecret(doc))
self._author = author
self.passphrase = os.environ.get(ENV_PASSPHRASE).encode()
self.salt = os.environ.get(ENV_SALT).encode()
def __iter__(self):
"""
Make the secret management object iterable
:return: the wrapped documents
"""
return (doc.pegleg_document for doc in self.documents)
@staticmethod
def __check_environment():
"""
@ -81,7 +100,7 @@ class PeglegSecretManagement(object):
'Environment variable {} is not defined or '
'is an empty string.'.format(ENV_SALT))
def encrypt_secrets(self, save_path, author):
def encrypt_secrets(self, save_path):
"""
Wrap and encrypt the secrets documents included in the input file,
into pegleg manage secrets documents, and write the result in
@ -97,11 +116,34 @@ class PeglegSecretManagement(object):
:type author: string
"""
doc_list, encrypted_docs = self.get_encrypted_secrets()
if encrypted_docs:
files.write(save_path, doc_list)
click.echo('Wrote encrypted data to: {}'.format(save_path))
else:
LOG.debug('All documents in file: {} are either already encrypted '
'or have cleartext storage policy. '
'Skipping.'.format(self.file_path))
def get_encrypted_secrets(self):
"""
:return doc_list: The list of documents
:rtype doc_list: list
:return encrypted_docs: Whether any documents were encrypted
:rtype encrypted_docs: bool
"""
if self._generated and not self._author:
raise ValueError("An author is needed to encrypt "
"generated documents. "
"Specify it when PeglegSecretManagement "
"is initialized.")
encrypted_docs = False
doc_list = []
for doc in self.documents:
# do not re-encrypt already encrypted data
if doc.is_encrypted():
doc_list.append(doc)
continue
# only encrypt if storagePolicy is set to encrypted.
@ -113,16 +155,10 @@ class PeglegSecretManagement(object):
doc.set_secret(
encrypt(doc.get_secret().encode(), self.passphrase, self.salt))
doc.set_encrypted(author)
doc.set_encrypted(self._author)
encrypted_docs = True
doc_list.append(doc.pegleg_document)
if encrypted_docs:
files.write(save_path, doc_list)
LOG.info('Wrote data to: {}.'.format(save_path))
else:
LOG.debug('All documents in file: {} are either already encrypted '
'or have cleartext storage policy. '
'Skipping.'.format(self.file_path))
return doc_list, encrypted_docs
def decrypt_secrets(self):
"""Decrypt and unwrap pegleg managed encrypted secrets documents

View File

@ -6,5 +6,6 @@ cryptography==2.3.1
python-dateutil==2.7.3
# External dependencies
rstr==2.2.6
git+https://github.com/openstack/airship-deckhand.git@7d697012fcbd868b14670aa9cf895acfad5a7f8d
git+https://github.com/openstack/airship-shipyard.git@44f7022df6438de541501c2fdd5c46df198b82bf#egg=shipyard_client&subdirectory=src/bin/shipyard_client

View File

@ -31,8 +31,9 @@ setup(
'pegleg=pegleg.cli:main',
]},
include_package_data=True,
package_dir={'pegleg': 'pegleg'},
package_data={
'schemas': [
'pegleg': [
'schemas/*.yaml',
],
},

View File

@ -0,0 +1,212 @@
---
# The purpose of this file is to define the Passpharase certificates for the environment
#
schema: pegleg/PassphraseCatalog/v1
metadata:
schema: metadata/Document/v1
name: cluster-passphrases
layeringDefinition:
abstract: false
layer: site
storagePolicy: cleartext
data:
passphrases:
- description: 'short description of the passphrase'
document_name: ceph_swift_keystone_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_keystone_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_armada_keystone_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_postgres_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_oslo_db_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_deckhand_postgres_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_deckhand_keystone_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_barbican_keystone_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_barbican_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_oslo_messaging_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_drydock_postgres_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_drydock_keystone_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_maas_postgres_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_keystone_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_promenade_keystone_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_shipyard_keystone_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_shipyard_postgres_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_airflow_postgres_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_rabbitmq_erlang_cookie
encrypted: true
- description: 'short description of the passphrase'
document_name: maas_region_secret
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_barbican_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_barbican_oslo_messaging_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_barbican_oslo_messaging_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_barbican_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_barbican_rabbitmq_erlang_cookie
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_cinder_oslo_messaging_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_cinder_oslo_messaging_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_cinder_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_cinder_rabbitmq_erlang_cookie
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_glance_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_glance_oslo_messaging_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_glance_oslo_messaging_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_glance_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_glance_rabbitmq_erlang_cookie
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_heat_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_heat_oslo_messaging_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_heat_oslo_messaging_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_heat_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_heat_rabbitmq_erlang_cookie
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_heat_stack_user_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_heat_trustee_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_horizon_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_infra_elasticsearch_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_infra_grafana_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_infra_grafana_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_infra_grafana_oslo_db_session_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_infra_kibana_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_infra_openstack_exporter_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_infra_oslo_db_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_keystone_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_keystone_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_keystone_oslo_messaging_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_keystone_oslo_messaging_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_keystone_rabbitmq_erlang_cookie
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_neutron_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_neutron_oslo_messaging_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_neutron_oslo_messaging_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_neutron_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_neutron_rabbitmq_erlang_cookie
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_nova_oslo_db_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_nova_oslo_messaging_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_nova_oslo_messaging_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_nova_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_nova_rabbitmq_erlang_cookie
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_oslo_db_admin_password
encrypted: true
- description: 'short description of the passphrase'
document_name: osh_placement_password
encrypted: true
...

View File

@ -0,0 +1,178 @@
# 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
import tempfile
import mock
import string
import yaml
from pegleg.engine.util.passphrase import Passphrase
from pegleg.engine.generators.passpharase_generator import PassphraseGenerator
from pegleg.engine.util import encryption
from pegleg.engine import util
import pegleg
from pegleg.engine.util.pegleg_secret_management import ENV_PASSPHRASE
from pegleg.engine.util.pegleg_secret_management import ENV_SALT
TEST_PASSPHRASES_CATALOG = yaml.load("""
---
schema: pegleg/PassphraseCatalog/v1
metadata:
schema: metadata/Document/v1
name: cluster-passphrases
layeringDefinition:
abstract: false
layer: site
storagePolicy: cleartext
data:
passphrases:
- description: 'short description of the passphrase'
document_name: ceph_swift_keystone_password
encrypted: true
- description: 'short description of the passphrase'
document_name: ucp_keystone_admin_password
encrypted: true
length: 24
- description: 'short description of the passphrase'
document_name: osh_barbican_oslo_db_password
encrypted: true
length: 23
- description: 'short description of the passphrase'
document_name: osh_cinder_password
encrypted: true
length: 25
- description: 'short description of the passphrase'
document_name: osh_oslo_db_admin_password
encrypted: true
length: 0
- description: 'short description of the passphrase'
document_name: osh_placement_password
encrypted: true
length: 32
...
""")
TEST_REPOSITORIES = {
'repositories': {
'global': {
'revision': '843d1a50106e1f17f3f722e2ef1634ae442fe68f',
'url': 'ssh://REPO_USERNAME@gerrit:29418/aic-clcp-manifests.git'
},
'secrets': {
'revision': 'master',
'url': ('ssh://REPO_USERNAME@gerrit:29418/aic-clcp-security-'
'manifests.git')
}
}
}
TEST_SITE_DEFINITION = {
'data': {
'revision': 'v1.0',
'site_type': 'cicd',
},
'metadata': {
'layeringDefinition': {
'abstract': 'false',
'layer': 'site',
},
'name': 'test-site',
'schema': 'metadata/Document/v1',
'storagePolicy': 'cleartext',
},
'schema': 'pegleg/SiteDefinition/v1',
}
TEST_SITE_DOCUMENTS = [TEST_SITE_DEFINITION, TEST_PASSPHRASES_CATALOG]
def test_passphrase_default_len():
p_util = Passphrase()
passphrase = p_util.get_pass()
assert len(passphrase) == 24
alphabet = set(string.punctuation + string.ascii_letters + string.digits)
assert any(c in alphabet for c in passphrase)
def test_passphrase_short_len():
p_util = Passphrase()
p = p_util.get_pass(0)
assert len(p) == 24
p = p_util.get_pass(23)
assert len(p) == 24
p = p_util.get_pass(-1)
assert len(p) == 24
def test_passphrase_long_len():
p_util = Passphrase()
p = p_util.get_pass(25)
assert len(p) == 25
p = p_util.get_pass(128)
assert len(p) == 128
@mock.patch.object(
util.definition,
'documents_for_site',
autospec=True,
return_value=TEST_SITE_DOCUMENTS)
@mock.patch.object(
pegleg.config,
'get_site_repo',
autospec=True,
return_value='cicd_site_repo')
@mock.patch.object(
util.definition,
'site_files',
autospec=True,
return_value=[
'cicd_site_repo/site/cicd/passphrases/passphrase-catalog.yaml', ])
@mock.patch.dict(os.environ, {
ENV_PASSPHRASE: 'ytrr89erARAiPE34692iwUMvWqqBvC',
ENV_SALT: 'MySecretSalt'})
def test_generate_passphrases(*_):
dir = tempfile.mkdtemp()
os.makedirs(os.path.join(dir, 'cicd_site_repo'), exist_ok=True)
PassphraseGenerator('cicd', dir, 'test_author').generate()
for passphrase in TEST_PASSPHRASES_CATALOG['data']['passphrases']:
passphrase_file_name = '{}.yaml'.format(passphrase['document_name'])
passphrase_file_path = os.path.join(dir, 'site', 'cicd', 'secrets',
'passphrases',
passphrase_file_name)
assert os.path.isfile(passphrase_file_path)
with open(passphrase_file_path) as stream:
doc = yaml.load(stream)
assert doc['schema'] == 'pegleg/PeglegManagedDocument/v1'
assert doc['metadata']['storagePolicy'] == 'cleartext'
assert 'encrypted' in doc['data']
assert doc['data']['encrypted']['by'] == 'test_author'
assert 'generated' in doc['data']
assert doc['data']['generated']['by'] == 'test_author'
assert 'managedDocument' in doc['data']
assert doc['data']['managedDocument']['metadata'][
'storagePolicy'] == 'encrypted'
decrypted_passphrase = encryption.decrypt(
doc['data']['managedDocument']['data'],
os.environ['PEGLEG_PASSPHRASE'].encode(),
os.environ['PEGLEG_SALT'].encode())
if passphrase_file_name == 'osh_placement_password.yaml':
assert len(decrypted_passphrase) == 32
elif passphrase_file_name == 'osh_cinder_password.yaml':
assert len(decrypted_passphrase) == 25
else:
assert len(decrypted_passphrase) == 24

View File

@ -33,8 +33,10 @@ from pegleg.engine.util.pegleg_secret_management import ENV_PASSPHRASE
from pegleg.engine.util.pegleg_secret_management import ENV_SALT
from pegleg.engine.util.pegleg_secret_management import PeglegSecretManagement
from tests.unit import test_utils
from tests.unit.fixtures import temp_path, create_tmp_deployment_files, _gen_document
from tests.unit.test_cli import TestSiteSecretsActions, BaseCLIActionTest, TEST_PARAMS
from tests.unit.fixtures import temp_path, create_tmp_deployment_files, \
_gen_document
from tests.unit.test_cli import TestSiteSecretsActions, BaseCLIActionTest, \
TEST_PARAMS
TEST_DATA = """
---
@ -69,10 +71,9 @@ def test_encrypt_and_decrypt():
ENV_SALT: 'MySecretSalt'
})
def test_short_passphrase():
with pytest.raises(
click.ClickException,
match=r'.*is not at least 24-character long.*'):
PeglegSecretManagement('file_path')
with pytest.raises(click.ClickException,
match=r'.*is not at least 24-character long.*'):
PeglegSecretManagement(file_path='file_path', author='test_author')
@mock.patch.dict(os.environ, {
@ -129,6 +130,26 @@ def test_pegleg_secret_management_constructor_with_invalid_arguments():
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, {
@ -143,14 +164,19 @@ def test_encrypt_decrypt_using_file_path(temp_path):
save_path = os.path.join(temp_path, 'encrypted_secrets_file.yaml')
# encrypt documents and validate that they were encrypted
doc_mgr = PeglegSecretManagement(file_path=file_path)
doc_mgr.encrypt_secrets(save_path, 'test_author')
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(save_path)
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']
@ -166,8 +192,9 @@ def test_encrypt_decrypt_using_docs(temp_path):
save_path = os.path.join(temp_path, 'encrypted_secrets_file.yaml')
# encrypt documents and validate that they were encrypted
doc_mgr = PeglegSecretManagement(docs=test_data)
doc_mgr.encrypt_secrets(save_path, 'test_author')
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'
@ -177,7 +204,8 @@ def test_encrypt_decrypt_using_docs(temp_path):
encrypted_data = list(yaml.safe_load_all(stream))
# decrypt documents and validate that they were decrypted
doc_mgr = PeglegSecretManagement(docs=encrypted_data)
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']
@ -190,6 +218,10 @@ def test_encrypt_decrypt_using_docs(temp_path):
@pytest.mark.skipif(
not pki_utility.PKIUtility.cfssl_exists(),
reason='cfssl must be installed to execute these tests')
@mock.patch.dict(os.environ, {
ENV_PASSPHRASE: 'ytrr89erARAiPE34692iwUMvWqqBvC',
ENV_SALT: 'MySecretSalt'
})
def test_generate_pki_using_local_repo_path(create_tmp_deployment_files):
"""Validates ``generate-pki`` action using local repo path."""
# Scenario:
@ -212,6 +244,10 @@ def test_generate_pki_using_local_repo_path(create_tmp_deployment_files):
@pytest.mark.skipif(
not pki_utility.PKIUtility.cfssl_exists(),
reason='cfssl must be installed to execute these tests')
@mock.patch.dict(os.environ, {
ENV_PASSPHRASE: 'ytrr89erARAiPE34692iwUMvWqqBvC',
ENV_SALT: 'MySecretSalt'
})
def test_check_expiry(create_tmp_deployment_files):
""" Validates check_expiry """
repo_path = str(git.git_handler(TEST_PARAMS["repo_url"],
@ -228,9 +264,11 @@ def test_check_expiry(create_tmp_deployment_files):
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['data']['managedDocument']['schema'] == \
if result['schema'] == \
"deckhand/Certificate/v1":
cert = result['data']['managedDocument']['data']
cert = result['data']
assert not pki_util.check_expiry(cert), \
"%s is expired!" % generated_file.name

View File

@ -28,7 +28,6 @@ from pegleg.engine.util import git
from tests.unit import test_utils
from tests.unit.fixtures import temp_path
TEST_PARAMS = {
"site_name": "airship-seaworthy",
"site_type": "foundry",
@ -67,7 +66,7 @@ class BaseCLIActionTest(object):
cls.repo_rev = TEST_PARAMS["repo_rev"]
cls.repo_name = TEST_PARAMS["repo_name"]
cls.treasuremap_path = git.git_handler(TEST_PARAMS["repo_url"],
ref=TEST_PARAMS["repo_rev"])
ref=TEST_PARAMS["repo_rev"])
class TestSiteCLIOptions(BaseCLIActionTest):
@ -377,7 +376,8 @@ class TestSiteCliActions(BaseCLIActionTest):
with mock.patch('pegleg.cli.ShipyardHelper') as mock_obj:
result = self.runner.invoke(cli.site,
['-r', repo_path, 'upload', self.site_name])
['-r', repo_path, 'upload',
self.site_name])
assert result.exit_code == 0
mock_obj.assert_called_once()
@ -442,6 +442,14 @@ class TestRepoCliActions(BaseCLIActionTest):
class TestSiteSecretsActions(BaseCLIActionTest):
"""Tests site secrets-related CLI actions."""
@classmethod
def setup_class(cls):
super(TestSiteSecretsActions, cls).setup_class()
cls.runner = CliRunner(env={
"PEGLEG_PASSPHRASE": 'ytrr89erARAiPE34692iwUMvWqqBvC',
"PEGLEG_SALT": "MySecretSalt"
})
def _validate_generate_pki_action(self, result):
assert result.exit_code == 0
@ -455,7 +463,7 @@ class TestSiteSecretsActions(BaseCLIActionTest):
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" % filename
assert list(result), "%s file is empty" % generated_file
@pytest.mark.skipif(
not pki_utility.PKIUtility.cfssl_exists(),
@ -493,9 +501,9 @@ class TestSiteSecretsActions(BaseCLIActionTest):
not pki_utility.PKIUtility.cfssl_exists(),
reason='cfssl must be installed to execute these tests')
@mock.patch.dict(os.environ, {
"PEGLEG_PASSPHRASE": "123456789012345678901234567890",
"PEGLEG_SALT": "123456"
})
"PEGLEG_PASSPHRASE": "123456789012345678901234567890",
"PEGLEG_SALT": "123456"
})
def test_site_secrets_encrypt_local_repo_path(self):
"""Validates ``generate-pki`` action using local repo path."""
# Scenario:
@ -504,13 +512,15 @@ class TestSiteSecretsActions(BaseCLIActionTest):
repo_path = self.treasuremap_path
with open(os.path.join(repo_path, "site", "airship-seaworthy",
"secrets", "passphrases", "ceph_fsid.yaml"), "r") \
"secrets", "passphrases", "ceph_fsid.yaml"),
"r") \
as ceph_fsid_fi:
ceph_fsid = yaml.load(ceph_fsid_fi)
ceph_fsid["metadata"]["storagePolicy"] = "encrypted"
with open(os.path.join(repo_path, "site", "airship-seaworthy",
"secrets", "passphrases", "ceph_fsid.yaml"), "w") \
"secrets", "passphrases", "ceph_fsid.yaml"),
"w") \
as ceph_fsid_fi:
yaml.dump(ceph_fsid, ceph_fsid_fi)
@ -520,7 +530,8 @@ class TestSiteSecretsActions(BaseCLIActionTest):
assert result.exit_code == 0
with open(os.path.join(repo_path, "site", "airship-seaworthy",
"secrets", "passphrases", "ceph_fsid.yaml"), "r") \
"secrets", "passphrases", "ceph_fsid.yaml"),
"r") \
as ceph_fsid_fi:
ceph_fsid = yaml.load(ceph_fsid_fi)
assert "encrypted" in ceph_fsid["data"]