airship/pegleg
Code Issues Proposed changes

PKI Cert generation and check updates

Browse Source 
This patch:
1. Allows user to change valid duration of newly generated certs
default=1yr
2. Allows user to check certs that are expiring soon default=60d

Change-Id: Ia5c87a0c52b39b778f425599fa215fb67147c65b
This commit is contained in:
Alexander Hughes 2019-02-26 12:40:11 -06:00
parent 6348b83e3c
commit 7c52ab68da
9 changed files with 207 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
doc/source/cli/cli.rst
View File

@ -477,6 +477,14 @@ Dashes in the document names will be converted to underscores for consistency.
Name of site.
**days** (Optional).
Duration (in days) certificates should be valid. Default=365,
minimum=0, no maximum.
NOTE: A generated certificate where days = 0 should only be used for testing.
A certificate generated in such a way will be valid for 0 seconds.
Examples
""""""""
@ -492,10 +500,78 @@ Examples
secrets generate-pki \
<site_name> \
-o <output> \
-f <filename>
-f <filename> \
-d <days>
.. _command-line-repository-overrides:
Check PKI Certs
---------------
Determine if any PKI certificates from a site are expired, or will be expired
within N days (default N=60, no maximum, minimum 0). Print those cert names
and expiration dates to ``stdout``.
**-d / --days** (Optional).
Number of days past today's date to check certificate expirations.
Default days=60. Minimum days=0, days less than 0 will raise an exception.
No maximum days.
**site_name** (Required).
Name of the ``site``. The ``site_name`` must match a ``site`` name in the site
repository folder structure.
Usage:
::
./pegleg.sh site -r <site_repo> \
secrets check-pki-certs <site_name> <options>
Examples
^^^^^^^^
Example without days specified:
::
./pegleg.sh site -r <site_repo> secrets check-pki-certs <site_name>
Example with days specified:
::
./pegleg.sh site -r <site_repo> secrets check-pki-certs <site_name> -d <days>
Secrets
-------
A sub-group of site command group, which allows you to perform secrets
level operations for secrets documents of a site.
.. note::
For the CLI commands ``encrypt`` and ``decrypt`` in the ``secrets`` command
group, which encrypt or decrypt site secrets, two environment variables,
``PEGLEG_PASSPHRASE``, and ``PEGLEG_SALT``, are used to capture the
master passphrase, and the salt needed for encryption and decryption of the
site secrets. The contents of ``PEGLEG_PASSPHRASE``, and ``PEGLEG_SALT``
are not generated by Pegleg, but are created externally, and set by a
deployment engineers or tooling.
A minimum length of 24 for master passphrases will be checked by all CLI
commands, which use the ``PEGLEG_PASSPHRASE``. All other criteria around
master passphrase strength are assumed to be enforced elsewhere.
::
./pegleg.sh site -r <site_repo> -e <extra_repo> secrets <command> <options>
Encrypt
^^^^^^^

35
pegleg/cli.py
View File

@ -401,8 +401,15 @@ def secrets():
'for tracking provenance information in the PeglegManagedDocuments. '
'An attempt is made to automatically determine this value, '
'but should be provided.')
@click.option(
'-d',
'--days',
'days',
default=365,
help='Duration in days generated certificates should be valid. '
'Default is 365 days.')
@click.argument('site_name')
def generate_pki(site_name, author):
def generate_pki(site_name, author, days):
"""Generate certificates, certificate authorities and keypairs for a given
site.
@ -410,7 +417,8 @@ def generate_pki(site_name, author):
engine.repository.process_repositories(site_name,
overwrite_existing=True)
pkigenerator = catalog.pki_generator.PKIGenerator(site_name, author=author)
pkigenerator = catalog.pki_generator.PKIGenerator(
site_name, author=author, duration=days)
output_paths = pkigenerator.generate()
click.echo("Generated PKI files written to:\n%s" % '\n'.join(output_paths))
@ -509,6 +517,29 @@ def genesis_bundle(*, build_dir, validators, site_name):
site_name)
@secrets.command(
'check-pki-certs',
help='Determine if certificates in a sites PKICatalog are expired or '
'expiring within a specified number of days.')
@click.option(
'-d',
'--days',
'days',
default=60,
help='The number of days past today to check if certificates are valid.')
@click.argument('site_name')
def check_pki_certs(site_name, days):
"""Check PKI certificates of a site for expiration."""
engine.repository.process_repositories(site_name,
overwrite_existing=True)
cert_results = engine.secrets.check_cert_expiry(site_name, duration=days)
click.echo("The following certs will expire within {} days: \n{}"
.format(days, cert_results))
@main.group(help='Commands related to types')
@MAIN_REPOSITORY_OPTION
@REPOSITORY_CLONE_PATH_OPTION

8
pegleg/engine/catalog/pki_generator.py
View File

@ -44,9 +44,12 @@ class PKIGenerator(object):
"""
def __init__(self, sitename, block_strings=True, author=None):
def __init__(self, sitename, block_strings=True, author=None,
duration=365):
"""Constructor for ``PKIGenerator``.
:param int duration: Duration in days that generated certificates
are valid.
:param str sitename: Site name for which to retrieve documents used for
certificate and keypair generation.
:param bool block_strings: Whether to dump out certificate data as
@ -60,7 +63,8 @@ class PKIGenerator(object):
self._documents = util.definition.documents_for_site(sitename)
self._author = author
self.keys = pki_utility.PKIUtility(block_strings=block_strings)
self.keys = pki_utility.PKIUtility(block_strings=block_strings,
duration=duration)
self.outputs = collections.defaultdict(dict)
# Maps certificates to CAs in order to derive certificate paths.

36
pegleg/engine/catalog/pki_utility.py
View File

@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import datetime
import datetime
import json
import logging
import os
@ -25,11 +25,11 @@ from dateutil import parser
import pytz
import yaml
from pegleg.engine import exceptions
from pegleg.engine.util.pegleg_managed_document import \
PeglegManagedSecretsDocument
LOG = logging.getLogger(__name__)
_ONE_YEAR_IN_HOURS = '8760h' # 365 * 24
__all__ = ['PKIUtility']
@ -57,23 +57,27 @@ class PKIUtility(object):
except subprocess.CalledProcessError:
return False
def __init__(self, *, block_strings=True):
def __init__(self, *, block_strings=True, duration=None):
self.block_strings = block_strings
self._ca_config_string = None
self.duration = duration
@property
def ca_config(self):
if self.duration is not None and self.duration >= 0:
pass
else:
raise exceptions.PKICertificateInvalidDuration()
if not self._ca_config_string:
self._ca_config_string = json.dumps({
'signing': {
'default': {
# TODO(felipemonteiro): Make this configurable.
'expiry':
_ONE_YEAR_IN_HOURS,
str(24 * self.duration) + 'h',
'usages': [
'signing', 'key encipherment', 'server auth',
'client auth'
],
'client auth'],
},
},
})
@ -198,17 +202,27 @@ class PKIUtility(object):
"""Chek whether a given certificate is expired.
:param str cert: Client certificate that contains the public key.
:returns: True if certificate is expired, else False.
:rtype: bool
:returns: In dictionary format returns the expiration date of the cert
and True if the cert is or will be expired within the next
expire_in_days
:rtype: dict
"""
if self.duration is not None and self.duration >= 0:
pass
else:
raise exceptions.PKICertificateInvalidDuration()
info = self.cert_info(cert)
expiry_str = info['not_after']
expiry = parser.parse(expiry_str)
# expiry is timezone-aware; do the same for `now`.
now = pytz.utc.localize(datetime.utcnow())
return now > expiry
expiry_window = pytz.utc.localize(datetime.datetime.utcnow()) + \
datetime.timedelta(days=self.duration)
expired = expiry_window > expiry
expiry = expiry.strftime('%d-%b-%Y %H:%M:%S %Z')
return {'expiry_date': expiry, 'expired': expired}
def _cfssl(self, command, *, files=None):
"""Executes ``cfssl`` command via ``subprocess`` call."""

7
pegleg/engine/exceptions.py
View File

@ -99,10 +99,17 @@ class GenesisBundleGenerateException(PeglegBaseException):
message = 'Bundle generation failed on deckhand validation.'
class PKICertificateInvalidDuration(PeglegBaseException):
"""Exception for invalid duration of PKI Certificate."""
message = ('Provided duration is invalid. Certificate durations must be '
'a positive integer.')
#
# CREDENTIALS EXCEPTIONS
#
class PassphraseNotFoundException(PeglegBaseException):
"""Exception raised when passphrase is not set."""

38
pegleg/engine/secrets.py
View File

@ -16,6 +16,9 @@ import logging
import os
import yaml
from prettytable import PrettyTable
from pegleg.engine.catalog.pki_utility import PKIUtility
from pegleg.engine.generators.passphrase_generator import PassphraseGenerator
from pegleg.engine.util.cryptostring import CryptoString
from pegleg.engine.util import definition
@ -186,3 +189,38 @@ def wrap_secret(author, file_name, output_path, schema,
output_doc = managed_secret.pegleg_document
with open(output_path, "w") as output_fi:
yaml.safe_dump(output_doc, output_fi)
def check_cert_expiry(site_name, duration=60):
"""
Check certs from a sites PKICatalog to determine if they are expired or
expiring within N days
:param str site_name: The site to read from
:param int duration: Number of days from today to check cert
expirations
:rtype: str
"""
pki_util = PKIUtility(duration=duration)
# Create a table to output expired/expiring certs for this site.
cert_table = PrettyTable()
cert_table.field_names = ['cert_name', 'expiration_date']
s = definition.site_files(site_name)
for doc in s:
if 'certificate' in doc:
with open(doc, '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)
if cert_info['expired'] is True:
cert_table.add_row([doc, cert_info['expiry_date']])
# Return table of cert names and expiration dates that are expiring
return cert_table.get_string()

10
tests/unit/engine/catalog/test_pki_utility.py
View File

@ -91,7 +91,7 @@ class TestPKIUtility(object):
assert PRIVATE_KEY_HEADER in priv_key['data']
def test_generate_certificate(self):
pki_obj = pki_utility.PKIUtility()
pki_obj = pki_utility.PKIUtility(duration=365)
ca_cert_wrapper, ca_key_wrapper = pki_obj.generate_ca(
self.__class__.__name__)
ca_cert = ca_cert_wrapper['data']['managedDocument']
@ -121,7 +121,7 @@ class TestPKIUtility(object):
def test_check_expiry_is_expired_false(self):
"""Check that ``check_expiry`` returns False if cert isn't expired."""
pki_obj = pki_utility.PKIUtility()
pki_obj = pki_utility.PKIUtility(duration=0)
ca_config = json.loads(pki_obj.ca_config)
ca_config['signing']['default']['expiry'] = '1h'
@ -141,7 +141,7 @@ class TestPKIUtility(object):
cert = cert_wrapper['data']['managedDocument']
# Validate that the cert hasn't expired.
is_expired = pki_obj.check_expiry(cert=cert['data'])
is_expired = pki_obj.check_expiry(cert=cert['data'])['expired']
assert not is_expired
def test_check_expiry_is_expired_true(self):
@ -149,7 +149,7 @@ class TestPKIUtility(object):
Second values are used to demonstrate precision down to the second.
"""
pki_obj = pki_utility.PKIUtility()
pki_obj = pki_utility.PKIUtility(duration=0)
ca_config = json.loads(pki_obj.ca_config)
ca_config['signing']['default']['expiry'] = '1s'
@ -171,5 +171,5 @@ class TestPKIUtility(object):
time.sleep(2)
# Validate that the cert has expired.
is_expired = pki_obj.check_expiry(cert=cert['data'])
is_expired = pki_obj.check_expiry(cert=cert['data'])['expired']
assert is_expired

12
tests/unit/engine/test_secrets.py
View File

@ -237,7 +237,7 @@ def test_generate_pki_using_local_repo_path(create_tmp_deployment_files):
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(sitename=TEST_PARAMS["site_name"])
pki_generator = PKIGenerator(duration=365, sitename=TEST_PARAMS["site_name"])
generated_files = pki_generator.generate()
assert len(generated_files), 'No secrets were generated'
@ -259,10 +259,10 @@ def test_check_expiry(create_tmp_deployment_files):
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(sitename=TEST_PARAMS["site_name"])
pki_generator = PKIGenerator(duration=365, sitename=TEST_PARAMS["site_name"])
generated_files = pki_generator.generate()
pki_util = pki_utility.PKIUtility()
pki_util = pki_utility.PKIUtility(duration=0)
assert len(generated_files), 'No secrets were generated'
for generated_file in generated_files:
@ -276,5 +276,7 @@ def test_check_expiry(create_tmp_deployment_files):
if result['schema'] == \
"deckhand/Certificate/v1":
cert = result['data']
assert not pki_util.check_expiry(cert), \
"%s is expired!" % generated_file.name
cert_info = pki_util.check_expiry(cert)
assert cert_info['expired'] is False, \
"%s is expired/expiring on %s" % \
(generated_file.name, cert_info['expiry_date'])

10
tests/unit/test_cli.py
View File

@ -562,6 +562,15 @@ class TestSiteSecretsActions(BaseCLIActionTest):
result = self.runner.invoke(cli.site, ['-r', repo_path] + secrets_opts)
assert result.exit_code == 0, result.output
@pytest.mark.skipif(
not pki_utility.PKIUtility.cfssl_exists(),
reason='cfssl must be installed to execute these tests')
def test_check_pki_certs(self):
repo_path = self.treasuremap_path
secrets_opts = ['secrets', 'check-pki-certs', self.site_name]
result = self.runner.invoke(cli.site, ['-r', repo_path] + secrets_opts)
assert result.exit_code == 0, result.output
@mock.patch.dict(os.environ, {
"PEGLEG_PASSPHRASE": "123456789012345678901234567890",
"PEGLEG_SALT": "123456"
@ -608,7 +617,6 @@ class TestSiteSecretsActions(BaseCLIActionTest):
assert "encrypted" in doc["data"]
assert "managedDocument" in doc["data"]
class TestTypeCliActions(BaseCLIActionTest):
"""Tests type-level CLI actions."""