Infra/lemur
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'master' into hshafagh-src-dst-register

Browse Source
This commit is contained in:
Hossein Shafagh
2019-04-09 20:52:33 -07:00
committed by GitHub
parent e139b92b24 70c8c2b5fc
commit bc8c7e114a
36 changed files with 822 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
lemur/__init__.py
View File

@ -62,8 +62,8 @@ LEMUR_BLUEPRINTS = (
)
def create_app(config=None):
app = factory.create_app(app_name=__name__, blueprints=LEMUR_BLUEPRINTS, config=config)
def create_app(config_path=None):
app = factory.create_app(app_name=__name__, blueprints=LEMUR_BLUEPRINTS, config=config_path)
configure_hook(app)
return app

8
lemur/certificates/models.py
View File

@ -101,7 +101,7 @@ class Certificate(db.Model):
issuer = Column(String(128))
serial = Column(String(128))
cn = Column(String(128))
deleted = Column(Boolean, index=True)
deleted = Column(Boolean, index=True, default=False)
dns_provider_id = Column(Integer(), ForeignKey('dns_providers.id', ondelete='CASCADE'), nullable=True)
not_before = Column(ArrowType)
@ -192,12 +192,16 @@ class Certificate(db.Model):
def check_integrity(self):
"""
Integrity checks: Does the cert have a matching private key?
Integrity checks: Does the cert have a valid chain and matching private key?
"""
if self.private_key:
validators.verify_private_key_match(utils.parse_private_key(self.private_key), self.parsed_cert,
error_class=AssertionError)
if self.chain:
chain = [self.parsed_cert] + utils.parse_cert_chain(self.chain)
validators.verify_cert_chain(chain, error_class=AssertionError)
@cached_property
def parsed_cert(self):
assert self.body, "Certificate body not set"

23
lemur/certificates/schemas.py
View File

@ -10,6 +10,7 @@ from marshmallow import fields, validate, validates_schema, post_load, pre_load
from marshmallow.exceptions import ValidationError
from lemur.authorities.schemas import AuthorityNestedOutputSchema
from lemur.certificates import utils as cert_utils
from lemur.common import missing, utils, validators
from lemur.common.fields import ArrowDateTime, Hex
from lemur.common.schema import LemurInputSchema, LemurOutputSchema
@ -96,6 +97,9 @@ class CertificateInputSchema(CertificateCreationSchema):
@validates_schema
def validate_authority(self, data):
if isinstance(data['authority'], str):
raise ValidationError("Authority not found.")
if not data['authority'].active:
raise ValidationError("The authority is inactive.", ['authority'])
@ -107,6 +111,11 @@ class CertificateInputSchema(CertificateCreationSchema):
def load_data(self, data):
if data.get('replacements'):
data['replaces'] = data['replacements'] # TODO remove when field is deprecated
if data.get('csr'):
dns_names = cert_utils.get_dns_names_from_csr(data['csr'])
if not data['extensions']['subAltNames']['names']:
data['extensions']['subAltNames']['names'] = []
data['extensions']['subAltNames']['names'] += dns_names
return missing.convert_validity_years(data)
@ -245,8 +254,7 @@ class CertificateUploadInputSchema(CertificateCreationSchema):
external_id = fields.String(missing=None, allow_none=True)
private_key = fields.String()
body = fields.String(required=True)
chain = fields.String(validate=validators.public_certificate, missing=None,
allow_none=True) # TODO this could be multiple certificates
chain = fields.String(missing=None, allow_none=True)
destinations = fields.Nested(AssociatedDestinationSchema, missing=[], many=True)
notifications = fields.Nested(AssociatedNotificationSchema, missing=[], many=True)
@ -260,7 +268,7 @@ class CertificateUploadInputSchema(CertificateCreationSchema):
raise ValidationError('Destinations require private key.')
@validates_schema
def validate_cert_private_key(self, data):
def validate_cert_private_key_chain(self, data):
cert = None
key = None
if data.get('body'):
@ -279,6 +287,15 @@ class CertificateUploadInputSchema(CertificateCreationSchema):
# Throws ValidationError
validators.verify_private_key_match(key, cert)
if data.get('chain'):
try:
chain = utils.parse_cert_chain(data['chain'])
except ValueError:
raise ValidationError("Invalid certificate in certificate chain.", field_names=['chain'])
# Throws ValidationError
validators.verify_cert_chain([cert] + chain)
class CertificateExportInputSchema(LemurInputSchema):
plugin = fields.Nested(PluginInputSchema)

23
lemur/certificates/service.py
View File

@ -54,7 +54,7 @@ def get_by_name(name):
def get_by_serial(serial):
"""
Retrieves certificate by it's Serial.
Retrieves certificate(s) by serial number.
:param serial:
:return:
"""
@ -64,6 +64,22 @@ def get_by_serial(serial):
return Certificate.query.filter(Certificate.serial == serial).all()
def get_by_attributes(conditions):
"""
Retrieves certificate(s) by conditions given in a hash of given key=>value pairs.
:param serial:
:return:
"""
# Ensure that each of the given conditions corresponds to actual columns
# if not, silently remove it
for attr in conditions.keys():
if attr not in Certificate.__table__.columns:
conditions.pop(attr)
query = database.session_query(Certificate)
return database.find_all(query, Certificate, conditions).all()
def delete(cert_id):
"""
Delete's a certificate.
@ -301,7 +317,7 @@ def render(args):
if filt:
terms = filt.split(';')
term = '{0}%'.format(terms[1])
term = '%{0}%'.format(terms[1])
# Exact matches for quotes. Only applies to name, issuer, and cn
if terms[1].startswith('"') and terms[1].endswith('"'):
term = terms[1][1:-1]
@ -365,6 +381,9 @@ def render(args):
now = arrow.now().format('YYYY-MM-DD')
query = query.filter(Certificate.not_after <= to).filter(Certificate.not_after >= now)
if current_app.config.get('ALLOW_CERT_DELETION', False):
query = query.filter(Certificate.deleted == False) # noqa
result = database.sort_and_page(query, Certificate, args)
return result

42
lemur/certificates/utils.py Normal file
View File

@ -0,0 +1,42 @@
"""
Utils to parse certificate data.
.. module: lemur.certificates.hooks
:platform: Unix
:copyright: (c) 2019 by Javier Ramos, see AUTHORS for more
:license: Apache, see LICENSE for more details.
.. moduleauthor:: Javier Ramos <javier.ramos@booking.com>
"""
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from marshmallow.exceptions import ValidationError
def get_dns_names_from_csr(data):
"""
Fetches DNSNames from CSR.
Potentially extendable to any kind of SubjectAlternativeName
:param data: PEM-encoded string with CSR
:return:
"""
dns_names = []
try:
request = x509.load_pem_x509_csr(data.encode('utf-8'), default_backend())
except Exception:
raise ValidationError('CSR presented is not valid.')
try:
alt_names = request.extensions.get_extension_for_class(x509.SubjectAlternativeName)
for name in alt_names.value.get_values_for_type(x509.DNSName):
dns_name = {
'nameType': 'DNSName',
'value': name
}
dns_names.append(dns_name)
except x509.ExtensionNotFound:
pass
return dns_names

19
lemur/certificates/views.py
View File

@ -6,10 +6,9 @@
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
"""
import base64
import arrow
from builtins import str
from flask import Blueprint, make_response, jsonify, g
from flask import Blueprint, make_response, jsonify, g, current_app
from flask_restful import reqparse, Api, inputs
from lemur.common.schema import validate_schema
@ -678,19 +677,26 @@ class Certificates(AuthenticatedResource):
.. sourcecode:: http
HTTP/1.1 200 OK
HTTP/1.1 204 OK
:reqheader Authorization: OAuth token to authenticate
:statuscode 204: no error
:statuscode 403: unauthenticated
:statusoode 404: certificate not found
:statuscode 404: certificate not found
:statuscode 405: certificate deletion is disabled
"""
if not current_app.config.get('ALLOW_CERT_DELETION', False):
return dict(message="Certificate deletion is disabled"), 405
cert = service.get(certificate_id)
if not cert:
return dict(message="Cannot find specified certificate"), 404
if cert.deleted:
return dict(message="Certificate is already deleted"), 412
# allow creators
if g.current_user != cert.user:
owner_role = role_service.get_by_name(cert.owner)
@ -699,12 +705,9 @@ class Certificates(AuthenticatedResource):
if not permission.can():
return dict(message='You are not authorized to delete this certificate'), 403
if arrow.get(cert.not_after) > arrow.utcnow():
return dict(message='Certificate is still valid, only expired certificates can be deleted'), 412
service.update(certificate_id, deleted=True)
log_service.create(g.current_user, 'delete_cert', certificate=cert)
return '', 204
return 'Certificate deleted', 204
class NotificationCertificatesList(AuthenticatedResource):

31
lemur/common/celery.py
View File

@ -47,6 +47,19 @@ def make_celery(app):
celery = make_celery(flask_app)
def is_task_active(fun, task_id, args):
from celery.task.control import inspect
i = inspect()
active_tasks = i.active()
for _, tasks in active_tasks.items():
for task in tasks:
if task.get("id") == task_id:
continue
if task.get("name") == fun and task.get("args") == str(args):
return True
return False
@celery.task()
def fetch_acme_cert(id):
"""
@ -224,5 +237,21 @@ def sync_source(source):
:param source:
:return:
"""
current_app.logger.debug("Syncing source {}".format(source))
function = "{}.{}".format(__name__, sys._getframe().f_code.co_name)
task_id = celery.current_task.request.id
log_data = {
"function": function,
"message": "Syncing source",
"source": source,
"task_id": task_id,
}
current_app.logger.debug(log_data)
if is_task_active(function, task_id, (source,)):
log_data["message"] = "Skipping task: Task is already active"
current_app.logger.debug(log_data)
return
sync([source])
log_data["message"] = "Done syncing source"
current_app.logger.debug(log_data)

13
lemur/common/defaults.py
View File

@ -3,6 +3,8 @@ import unicodedata
from cryptography import x509
from flask import current_app
from lemur.common.utils import is_selfsigned
from lemur.extensions import sentry
from lemur.constants import SAN_NAMING_TEMPLATE, DEFAULT_NAMING_TEMPLATE
@ -229,15 +231,22 @@ def issuer(cert):
"""
Gets a sane issuer slug from a given certificate, stripping non-alphanumeric characters.
:param cert:
For self-signed certificates, the special value '<selfsigned>' is returned.
If issuer cannot be determined, '<unknown>' is returned.
:param cert: Parsed certificate object
:return: Issuer slug
"""
# If certificate is self-signed, we return a special value -- there really is no distinct "issuer" for it
if is_selfsigned(cert):
return '<selfsigned>'
# Try Common Name or fall back to Organization name
attrs = (cert.issuer.get_attributes_for_oid(x509.OID_COMMON_NAME) or
cert.issuer.get_attributes_for_oid(x509.OID_ORGANIZATION_NAME))
if not attrs:
current_app.logger.error("Unable to get issuer! Cert serial {:x}".format(cert.serial_number))
return "Unknown"
return '<unknown>'
return text_to_slug(attrs[0].value, '')

60
lemur/common/utils.py
View File

@ -7,13 +7,15 @@
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
"""
import random
import re
import string
import sqlalchemy
from cryptography import x509
from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa, ec
from cryptography.hazmat.primitives.asymmetric import rsa, ec, padding
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from flask_restful.reqparse import RequestParser
from sqlalchemy import and_, func
@ -66,6 +68,26 @@ def parse_private_key(private_key):
return load_pem_private_key(private_key.encode('utf8'), password=None, backend=default_backend())
def split_pem(data):
"""
Split a string of several PEM payloads to a list of strings.
:param data: String
:return: List of strings
"""
return re.split("\n(?=-----BEGIN )", data)
def parse_cert_chain(pem_chain):
"""
Helper function to split and parse a series of PEM certificates.
:param pem_chain: string
:return: List of parsed certificates
"""
return [parse_certificate(cert) for cert in split_pem(pem_chain) if pem_chain]
def parse_csr(csr):
"""
Helper function that parses a CSR.
@ -143,6 +165,42 @@ def generate_private_key(key_type):
)
def check_cert_signature(cert, issuer_public_key):
"""
Check a certificate's signature against an issuer public key.
Before EC validation, make sure we support the algorithm, otherwise raise UnsupportedAlgorithm
On success, returns None; on failure, raises UnsupportedAlgorithm or InvalidSignature.
"""
if isinstance(issuer_public_key, rsa.RSAPublicKey):
# RSA requires padding, just to make life difficult for us poor developers :(
if cert.signature_algorithm_oid == x509.SignatureAlgorithmOID.RSASSA_PSS:
# In 2005, IETF devised a more secure padding scheme to replace PKCS #1 v1.5. To make sure that
# nobody can easily support or use it, they mandated lots of complicated parameters, unlike any
# other X.509 signature scheme.
# https://tools.ietf.org/html/rfc4056
raise UnsupportedAlgorithm("RSASSA-PSS not supported")
else:
padder = padding.PKCS1v15()
issuer_public_key.verify(cert.signature, cert.tbs_certificate_bytes, padder, cert.signature_hash_algorithm)
elif isinstance(issuer_public_key, ec.EllipticCurvePublicKey) and isinstance(ec.ECDSA(cert.signature_hash_algorithm), ec.ECDSA):
issuer_public_key.verify(cert.signature, cert.tbs_certificate_bytes, ec.ECDSA(cert.signature_hash_algorithm))
else:
raise UnsupportedAlgorithm("Unsupported Algorithm '{var}'.".format(var=cert.signature_algorithm_oid._name))
def is_selfsigned(cert):
"""
Returns True if the certificate is self-signed.
Returns False for failed verification or unsupported signing algorithm.
"""
try:
check_cert_signature(cert, cert.public_key())
# If verification was successful, it's self-signed.
return True
except InvalidSignature:
return False
def is_weekend(date):
"""
Determines if a given date is on a weekend.

48
lemur/common/validators.py
View File

@ -1,27 +1,14 @@
import re
from cryptography import x509
from cryptography.exceptions import UnsupportedAlgorithm, InvalidSignature
from cryptography.hazmat.backends import default_backend
from cryptography.x509 import NameOID
from flask import current_app
from marshmallow.exceptions import ValidationError
from lemur.auth.permissions import SensitiveDomainPermission
from lemur.common.utils import parse_certificate, is_weekend
def public_certificate(body):
"""
Determines if specified string is valid public certificate.
:param body:
:return:
"""
try:
parse_certificate(body)
except Exception as e:
current_app.logger.exception(e)
raise ValidationError('Public certificate presented is not valid.')
from lemur.common.utils import check_cert_signature, is_weekend
def common_name(value):
@ -138,3 +125,34 @@ def verify_private_key_match(key, cert, error_class=ValidationError):
"""
if key.public_key().public_numbers() != cert.public_key().public_numbers():
raise error_class("Private key does not match certificate.")
def verify_cert_chain(certs, error_class=ValidationError):
"""
Verifies that the certificates in the chain are correct.
We don't bother with full cert validation but just check that certs in the chain are signed by the next, to avoid
basic human errors -- such as pasting the wrong certificate.
:param certs: List of parsed certificates, use parse_cert_chain()
:param error_class: Exception class to raise on error
"""
cert = certs[0]
for issuer in certs[1:]:
# Use the current cert's public key to verify the previous signature.
# "certificate validation is a complex problem that involves much more than just signature checks"
try:
check_cert_signature(cert, issuer.public_key())
except InvalidSignature:
# Avoid circular import.
from lemur.common import defaults
raise error_class("Incorrect chain certificate(s) provided: '%s' is not signed by '%s'"
% (defaults.common_name(cert) or 'Unknown', defaults.common_name(issuer)))
except UnsupportedAlgorithm as err:
current_app.logger.warning("Skipping chain validation: %s", err)
# Next loop will validate that *this issuer* cert is signed by the next chain cert.
cert = issuer

4
lemur/manage.py
View File

@ -50,7 +50,7 @@ from lemur.pending_certificates.models import PendingCertificate # noqa
from lemur.dns_providers.models import DnsProvider # noqa
manager = Manager(create_app)
manager.add_option('-c', '--config', dest='config')
manager.add_option('-c', '--config', dest='config_path', required=False)
migrate = Migrate(create_app)
@ -391,7 +391,7 @@ class LemurServer(Command):
# run startup tasks on an app like object
validate_conf(current_app, REQUIRED_VARIABLES)
app.app_uri = 'lemur:create_app(config="{0}")'.format(current_app.config.get('CONFIG_PATH'))
app.app_uri = 'lemur:create_app(config_path="{0}")'.format(current_app.config.get('CONFIG_PATH'))
return app.run()

23
lemur/migrations/versions/318b66568358_.py Normal file
View File

@ -0,0 +1,23 @@
""" Set 'deleted' flag from null to false on all certificates once
Revision ID: 318b66568358
Revises: 9f79024fe67b
Create Date: 2019-02-05 15:42:25.477587
"""
# revision identifiers, used by Alembic.
revision = '318b66568358'
down_revision = '9f79024fe67b'
from alembic import op
def upgrade():
connection = op.get_bind()
# Delete duplicate entries
connection.execute('UPDATE certificates SET deleted = false WHERE deleted IS NULL')
def downgrade():
pass

5
lemur/plugins/lemur_acme/plugin.py
View File

@ -459,7 +459,10 @@ class ACMEIssuerPlugin(IssuerPlugin):
"pending_cert": entry["pending_cert"],
})
except (PollError, AcmeError, Exception) as e:
current_app.logger.error("Unable to resolve pending cert: {}".format(pending_cert), exc_info=True)
order_url = order.uri
current_app.logger.error(
"Unable to resolve pending cert: {}. "
"Check out {} for more information.".format(pending_cert, order_url), exc_info=True)
certs.append({
"cert": False,
"pending_cert": entry["pending_cert"],

4
lemur/plugins/lemur_java/plugin.py
View File

@ -15,6 +15,8 @@ from cryptography.fernet import Fernet
from lemur.utils import mktempfile, mktemppath
from lemur.plugins.bases import ExportPlugin
from lemur.plugins import lemur_java as java
from lemur.common.utils import parse_certificate
from lemur.common.defaults import common_name
def run_process(command):
@ -233,7 +235,7 @@ class JavaKeystoreExportPlugin(ExportPlugin):
if self.get_option('alias', options):
alias = self.get_option('alias', options)
else:
alias = "blah"
alias = common_name(parse_certificate(body))
with mktemppath() as jks_tmp:
create_keystore(body, chain, jks_tmp, key, alias, passphrase)

5
lemur/plugins/lemur_openssl/plugin.py
View File

@ -14,7 +14,8 @@ from flask import current_app
from lemur.utils import mktempfile, mktemppath
from lemur.plugins.bases import ExportPlugin
from lemur.plugins import lemur_openssl as openssl
from lemur.common.utils import get_psuedo_random_string
from lemur.common.utils import get_psuedo_random_string, parse_certificate
from lemur.common.defaults import common_name
def run_process(command):
@ -122,7 +123,7 @@ class OpenSSLExportPlugin(ExportPlugin):
if self.get_option('alias', options):
alias = self.get_option('alias', options)
else:
alias = "blah"
alias = common_name(parse_certificate(body))
type = self.get_option('type', options)

5
lemur/plugins/lemur_vault_dest/__init__.py Normal file
View File

@ -0,0 +1,5 @@
try:
VERSION = __import__('pkg_resources') \
.get_distribution(__name__).version
except Exception as e:
VERSION = 'unknown'

173
lemur/plugins/lemur_vault_dest/plugin.py Normal file
View File

@ -0,0 +1,173 @@
"""
.. module: lemur.plugins.lemur_vault_dest.plugin
:platform: Unix
:copyright: (c) 2019
:license: Apache, see LICENCE for more details.
Plugin for uploading certificates and private key as secret to hashi vault
that can be pulled down by end point nodes.
.. moduleauthor:: Christopher Jolley <chris@alwaysjolley.com>
"""
import hvac
from flask import current_app
from lemur.common.defaults import common_name
from lemur.common.utils import parse_certificate
from lemur.plugins.bases import DestinationPlugin
from cryptography import x509
from cryptography.hazmat.backends import default_backend
class VaultDestinationPlugin(DestinationPlugin):
"""Hashicorp Vault Destination plugin for Lemur"""
title = 'Vault'
slug = 'hashi-vault-destination'
description = 'Allow the uploading of certificates to Hashi Vault as secret'
author = 'Christopher Jolley'
author_url = 'https://github.com/alwaysjolley/lemur'
options = [
{
'name': 'vaultUrl',
'type': 'str',
'required': True,
'validation': '^https?://[a-zA-Z0-9.:-]+$',
'helpMessage': 'Valid URL to Hashi Vault instance'
},
{
'name': 'vaultKvApiVersion',
'type': 'select',
'value': '2',
'available': [
'1',
'2'
],
'required': True,
'helpMessage': 'Version of the Vault KV API to use'
},
{
'name': 'vaultAuthTokenFile',
'type': 'str',
'required': True,
'validation': '(/[^/]+)+',
'helpMessage': 'Must be a valid file path!'
},
{
'name': 'vaultMount',
'type': 'str',
'required': True,
'validation': '^\S+$',
'helpMessage': 'Must be a valid Vault secrets mount name!'
},
{
'name': 'vaultPath',
'type': 'str',
'required': True,
'validation': '^([a-zA-Z0-9_-]+/?)+$',
'helpMessage': 'Must be a valid Vault secrets path'
},
{
'name': 'objectName',
'type': 'str',
'required': False,
'validation': '[0-9a-zA-Z:_-]+',
'helpMessage': 'Name to bundle certs under, if blank use cn'
},
{
'name': 'bundleChain',
'type': 'select',
'value': 'cert only',
'available': [
'Nginx',
'Apache',
'no chain'
],
'required': True,
'helpMessage': 'Bundle the chain into the certificate'
}
]
def __init__(self, *args, **kwargs):
super(VaultDestinationPlugin, self).__init__(*args, **kwargs)
def upload(self, name, body, private_key, cert_chain, options, **kwargs):
"""
Upload certificate and private key
:param private_key:
:param cert_chain:
:return:
"""
cname = common_name(parse_certificate(body))
url = self.get_option('vaultUrl', options)
token_file = self.get_option('vaultAuthTokenFile', options)
mount = self.get_option('vaultMount', options)
path = self.get_option('vaultPath', options)
bundle = self.get_option('bundleChain', options)
obj_name = self.get_option('objectName', options)
api_version = self.get_option('vaultKvApiVersion', options)
with open(token_file, 'r') as file:
token = file.readline().rstrip('\n')
client = hvac.Client(url=url, token=token)
client.secrets.kv.default_kv_version = api_version
if obj_name:
path = '{0}/{1}'.format(path, obj_name)
else:
path = '{0}/{1}'.format(path, cname)
secret = get_secret(client, mount, path)
secret['data'][cname] = {}
if bundle == 'Nginx' and cert_chain:
secret['data'][cname]['crt'] = '{0}\n{1}'.format(body, cert_chain)
elif bundle == 'Apache' and cert_chain:
secret['data'][cname]['crt'] = body
secret['data'][cname]['chain'] = cert_chain
else:
secret['data'][cname]['crt'] = body
secret['data'][cname]['key'] = private_key
san_list = get_san_list(body)
if isinstance(san_list, list):
secret['data'][cname]['san'] = san_list
try:
client.secrets.kv.create_or_update_secret(
path=path, mount_point=mount, secret=secret['data']
)
except ConnectionError as err:
current_app.logger.exception(
"Exception uploading secret to vault: {0}".format(err), exc_info=True)
def get_san_list(body):
""" parse certificate for SAN names and return list, return empty list on error """
san_list = []
try:
byte_body = body.encode('utf-8')
cert = x509.load_pem_x509_certificate(byte_body, default_backend())
ext = cert.extensions.get_extension_for_oid(x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME)
san_list = ext.value.get_values_for_type(x509.DNSName)
except x509.extensions.ExtensionNotFound:
pass
finally:
return san_list
def get_secret(client, mount, path):
""" retreiive existing data from mount path and return dictionary """
result = {'data': {}}
try:
if client.secrets.kv.default_kv_version == '1':
result = client.secrets.kv.v1.read_secret(path=path, mount_point=mount)
else:
result = client.secrets.kv.v2.read_secret_version(path=path, mount_point=mount)
except ConnectionError:
pass
finally:
return result

1
lemur/plugins/lemur_vault_dest/tests/conftest.py Normal file
View File

@ -0,0 +1 @@
from lemur.tests.conftest import * # noqa

7
lemur/sources/service.py
View File

@ -116,7 +116,12 @@ def sync_certificates(source, user):
for certificate in certificates:
exists = False
if certificate.get('name'):
if certificate.get('search', None):
conditions = certificate.pop('search')
exists = certificate_service.get_by_attributes(conditions)
if not exists and certificate.get('name'):
result = certificate_service.get_by_name(certificate['name'])
if result:
exists = [result]

2
lemur/tests/conf.py
View File

@ -186,3 +186,5 @@ LDAP_BASE_DN = 'dc=example,dc=com'
LDAP_EMAIL_DOMAIN = 'example.com'
LDAP_REQUIRED_GROUP = 'Lemur Access'
LDAP_DEFAULT_ROLE = 'role1'
ALLOW_CERT_DELETION = True

10
lemur/tests/conftest.py
View File

@ -3,6 +3,8 @@ import os
import datetime
import pytest
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from flask import current_app
from flask_principal import identity_changed, Identity
@ -41,7 +43,7 @@ def app(request):
Creates a new Flask application for a test duration.
Uses application factory `create_app`.
"""
_app = create_app(os.path.dirname(os.path.realpath(__file__)) + '/conf.py')
_app = create_app(config_path=os.path.dirname(os.path.realpath(__file__)) + '/conf.py')
ctx = _app.app_context()
ctx.push()
@ -263,6 +265,12 @@ def cert_builder(private_key):
.not_valid_after(datetime.datetime(2040, 1, 1)))
@pytest.fixture
def selfsigned_cert(cert_builder, private_key):
# cert_builder uses the same cert public key as 'private_key'
return cert_builder.sign(private_key, hashes.SHA256(), default_backend())
@pytest.fixture(scope='function')
def aws_credentials():
os.environ['AWS_ACCESS_KEY_ID'] = 'testing'

1
lemur/tests/factories.py
View File

@ -140,6 +140,7 @@ class CACertificateFactory(CertificateFactory):
class InvalidCertificateFactory(CertificateFactory):
body = INVALID_CERT_STR
private_key = ''
chain = ''
class AuthorityFactory(BaseFactory):

111
lemur/tests/test_certificates.py
View File

@ -41,6 +41,89 @@ def test_get_or_increase_name(session, certificate):
assert get_or_increase_name('certificate1', int(serial, 16)) == 'certificate1-{}-1'.format(serial)
def test_get_all_certs(session, certificate):
from lemur.certificates.service import get_all_certs
assert len(get_all_certs()) > 1
def test_get_by_name(session, certificate):
from lemur.certificates.service import get_by_name
found = get_by_name(certificate.name)
assert found
def test_get_by_serial(session, certificate):
from lemur.certificates.service import get_by_serial
found = get_by_serial(certificate.serial)
assert found
def test_delete_cert(session):
from lemur.certificates.service import delete, get
from lemur.tests.factories import CertificateFactory
delete_this = CertificateFactory(name='DELETEME')
session.commit()
cert_exists = get(delete_this.id)
# it needs to exist first
assert cert_exists
delete(delete_this.id)
cert_exists = get(delete_this.id)
# then not exist after delete
assert not cert_exists
def test_get_by_attributes(session, certificate):
from lemur.certificates.service import get_by_attributes
# Should get one cert
certificate1 = get_by_attributes({
'name': 'SAN-san.example.org-LemurTrustUnittestsClass1CA2018-20171231-20471231'
})
# Should get one cert using multiple attrs
certificate2 = get_by_attributes({
'name': 'test-cert-11111111-1',
'cn': 'san.example.org'
})
# Should get multiple certs
multiple = get_by_attributes({
'cn': 'LemurTrust Unittests Class 1 CA 2018',
'issuer': 'LemurTrustUnittestsRootCA2018'
})
assert len(certificate1) == 1
assert len(certificate2) == 1
assert len(multiple) > 1
def test_find_duplicates(session):
from lemur.certificates.service import find_duplicates
cert = {
'body': SAN_CERT_STR,
'chain': INTERMEDIATE_CERT_STR
}
dups1 = find_duplicates(cert)
cert['chain'] = ''
dups2 = find_duplicates(cert)
assert len(dups1) > 0
assert len(dups2) > 0
def test_get_certificate_primitives(certificate):
from lemur.certificates.service import get_certificate_primitives
@ -429,7 +512,7 @@ def test_certificate_upload_schema_invalid_chain(client):
'owner': 'pwner@example.com',
}
data, errors = CertificateUploadInputSchema().load(data)
assert errors == {'chain': ['Public certificate presented is not valid.']}
assert errors == {'chain': ['Invalid certificate in certificate chain.']}
def test_certificate_upload_schema_wrong_pkey(client):
@ -444,6 +527,30 @@ def test_certificate_upload_schema_wrong_pkey(client):
assert errors == {'_schema': ['Private key does not match certificate.']}
def test_certificate_upload_schema_wrong_chain(client):
from lemur.certificates.schemas import CertificateUploadInputSchema
data = {
'owner': 'pwner@example.com',
'body': SAN_CERT_STR,
'chain': ROOTCA_CERT_STR,
}
data, errors = CertificateUploadInputSchema().load(data)
assert errors == {'_schema': ["Incorrect chain certificate(s) provided: 'san.example.org' is not signed by "
"'LemurTrust Unittests Root CA 2018'"]}
def test_certificate_upload_schema_wrong_chain_2nd(client):
from lemur.certificates.schemas import CertificateUploadInputSchema
data = {
'owner': 'pwner@example.com',
'body': SAN_CERT_STR,
'chain': INTERMEDIATE_CERT_STR + '\n' + SAN_CERT_STR,
}
data, errors = CertificateUploadInputSchema().load(data)
assert errors == {'_schema': ["Incorrect chain certificate(s) provided: 'LemurTrust Unittests Class 1 CA 2018' is "
"not signed by 'san.example.org'"]}
def test_create_basic_csr(client):
csr_config = dict(
common_name='example.com',
@ -654,7 +761,7 @@ def test_certificate_put_with_data(client, certificate, issuer_plugin):
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 403),
(VALID_ADMIN_HEADER_TOKEN, 412),
(VALID_ADMIN_HEADER_TOKEN, 204),
(VALID_ADMIN_API_TOKEN, 412),
('', 401)
])

14
lemur/tests/test_defaults.py
View File

@ -81,6 +81,13 @@ def test_create_name(client):
datetime(2015, 5, 12, 0, 0, 0),
False
) == 'xn--mnchen-3ya.de-VertrauenswurdigAutoritat-20150507-20150512'
assert certificate_name(
'selfie.example.org',
'<selfsigned>',
datetime(2015, 5, 7, 0, 0, 0),
datetime(2025, 5, 12, 13, 37, 0),
False
) == 'selfie.example.org-selfsigned-20150507-20250512'
def test_issuer(client, cert_builder, issuer_private_key):
@ -106,4 +113,9 @@ def test_issuer(client, cert_builder, issuer_private_key):
cert = (cert_builder
.issuer_name(x509.Name([]))
.sign(issuer_private_key, hashes.SHA256(), default_backend()))
assert issuer(cert) == 'Unknown'
assert issuer(cert) == '<unknown>'
def test_issuer_selfsigned(selfsigned_cert):
from lemur.common.defaults import issuer
assert issuer(selfsigned_cert) == '<selfsigned>'

20
lemur/tests/test_utils.py
View File

@ -1,5 +1,7 @@
import pytest
from lemur.tests.vectors import SAN_CERT, INTERMEDIATE_CERT, ROOTCA_CERT, EC_CERT_EXAMPLE, ECDSA_PRIME256V1_CERT, ECDSA_SECP384r1_CERT, DSA_CERT
def test_generate_private_key():
from lemur.common.utils import generate_private_key
@ -71,3 +73,21 @@ KFfxwrO1
-----END CERTIFICATE-----'''
authority_key = get_authority_key(test_cert)
assert authority_key == 'feacb541be81771293affa412d8dc9f66a3ebb80'
def test_is_selfsigned(selfsigned_cert):
from lemur.common.utils import is_selfsigned
assert is_selfsigned(selfsigned_cert) is True
assert is_selfsigned(SAN_CERT) is False
assert is_selfsigned(INTERMEDIATE_CERT) is False
# Root CA certificates are also technically self-signed
assert is_selfsigned(ROOTCA_CERT) is True
assert is_selfsigned(EC_CERT_EXAMPLE) is False
# selfsigned certs
assert is_selfsigned(ECDSA_PRIME256V1_CERT) is True
assert is_selfsigned(ECDSA_SECP384r1_CERT) is True
# unsupported algorithm (DSA)
with pytest.raises(Exception):
is_selfsigned(DSA_CERT)

96
lemur/tests/vectors.py
View File

@ -45,6 +45,7 @@ ssvobJ6Xe2D4cCVjUmsqtFEztMgdqgmlcWyGdUKeXdi7CMoeTb4uO+9qRQq46wYW
n7K1z+W0Kp5yhnnPAoOioAP4vjASDx3z3RnLaZvMmcO7YdCIwhE5oGV0
-----END CERTIFICATE-----
"""
ROOTCA_CERT = parse_certificate(ROOTCA_CERT_STR)
ROOTCA_KEY = """\
-----BEGIN RSA PRIVATE KEY-----
MIIEowIBAAKCAQEAvyVpe0tfIzri3l3PYH2r7hW86wKF58GLY+Ua52rEO5E3eXQq
@ -393,3 +394,98 @@ zm3Cn4Ul8DO26w9QS4fmZjmnPOZFXYMWoOR6osHzb62PWQ8FBMqXcdToBV2Q9Iw4
PiFAxlc0tVjlLqQ=
-----END CERTIFICATE REQUEST-----
"""
EC_CERT_STR = """
-----BEGIN CERTIFICATE-----
MIIDxzCCAq+gAwIBAgIIHsJeci1JWAkwDQYJKoZIhvcNAQELBQAwVDELMAkGA1UE
BhMCVVMxHjAcBgNVBAoTFUdvb2dsZSBUcnVzdCBTZXJ2aWNlczElMCMGA1UEAxMc
R29vZ2xlIEludGVybmV0IEF1dGhvcml0eSBHMzAeFw0xOTAyMTMxNTM1NTdaFw0x
OTA1MDgxNTM1MDBaMGgxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDYWxpZm9ybmlh
MRYwFAYDVQQHDA1Nb3VudGFpbiBWaWV3MRMwEQYDVQQKDApHb29nbGUgTExDMRcw
FQYDVQQDDA53d3cuZ29vZ2xlLmNvbTBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IA
BKwMlIbd4rAwf6eWoa6RrR2w0s5k1M40XOORPf96PByPmld+qhjRMLvA/xcAxdCR
XdcMfaX6EUr0Zw8CepitMB2jggFSMIIBTjATBgNVHSUEDDAKBggrBgEFBQcDATAO
BgNVHQ8BAf8EBAMCB4AwGQYDVR0RBBIwEIIOd3d3Lmdvb2dsZS5jb20waAYIKwYB
BQUHAQEEXDBaMC0GCCsGAQUFBzAChiFodHRwOi8vcGtpLmdvb2cvZ3NyMi9HVFNH
SUFHMy5jcnQwKQYIKwYBBQUHMAGGHWh0dHA6Ly9vY3NwLnBraS5nb29nL0dUU0dJ
QUczMB0GA1UdDgQWBBQLovm8GG0oG91gOGCL58YPNoAlejAMBgNVHRMBAf8EAjAA
MB8GA1UdIwQYMBaAFHfCuFCaZ3Z2sS3ChtCDoH6mfrpLMCEGA1UdIAQaMBgwDAYK
KwYBBAHWeQIFAzAIBgZngQwBAgIwMQYDVR0fBCowKDAmoCSgIoYgaHR0cDovL2Ny
bC5wa2kuZ29vZy9HVFNHSUFHMy5jcmwwDQYJKoZIhvcNAQELBQADggEBAKFbmNOA
e3pJ7UVI5EmkAMZgSDRdrsLHV6F7WluuyYCyE/HFpZjBd6y8xgGtYWcask6edwrq
zrcXNEN/GY34AYre0M+p0xAs+lKSwkrJd2sCgygmzsBFtGwjW6lhjm+rg83zPHhH
mQZ0ShUR1Kp4TvzXgxj44RXOsS5ZyDe3slGiG4aw/hl+igO8Y8JMvcv/Tpzo+V75
BkDAFmLRi08NayfeyCqK/TcRpzxKMKhS7jEHK8Pzu5P+FyFHKqIsobi+BA+psOix
5nZLhrweLdKNz387mE2lSSKzr7qeLGHSOMt+ajQtZio4YVyZqJvg4Y++J0n5+Rjw
MXp8GrvTfn1DQ+o=
-----END CERTIFICATE-----
"""
EC_CERT_EXAMPLE = parse_certificate(EC_CERT_STR)
ECDSA_PRIME256V1_CERT_STR = """
-----BEGIN CERTIFICATE-----
MIICUTCCAfYCCQCvH7H/e2nuiDAKBggqhkjOPQQDAjCBrzELMAkGA1UEBhMCVVMx
EzARBgNVBAgMCkNhbGlmb3JuaWExEjAQBgNVBAcMCUxvcyBHYXRvczEjMCEGA1UE
CgwaTGVtdXJUcnVzdCBFbnRlcnByaXNlcyBMdGQxJjAkBgNVBAsMHVVuaXR0ZXN0
aW5nIE9wZXJhdGlvbnMgQ2VudGVyMSowKAYDVQQDDCFMZW11clRydXN0IFVuaXR0
ZXN0cyBSb290IENBIDIwMTkwHhcNMTkwMjI2MTgxMTUyWhcNMjkwMjIzMTgxMTUy
WjCBrzELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNhbGlmb3JuaWExEjAQBgNVBAcM
CUxvcyBHYXRvczEjMCEGA1UECgwaTGVtdXJUcnVzdCBFbnRlcnByaXNlcyBMdGQx
JjAkBgNVBAsMHVVuaXR0ZXN0aW5nIE9wZXJhdGlvbnMgQ2VudGVyMSowKAYDVQQD
DCFMZW11clRydXN0IFVuaXR0ZXN0cyBSb290IENBIDIwMTkwWTATBgcqhkjOPQIB
BggqhkjOPQMBBwNCAAQsnAVUtpDCFMK/k9Chynu8BWRVUBUYbGQ9Q9xeLR60J4fD
uBt48YpTqg5RMZEclVknMReXqTmqphOBo37/YVdlMAoGCCqGSM49BAMCA0kAMEYC
IQDQZ6xfBiCTHxY4GM4+zLeG1iPBUSfIJOjkFNViFZY/XAIhAJYmrkVQb/YjWCdd
Vl89McYhmV4IV7WDgUmUhkUSFXgy
-----END CERTIFICATE-----
"""
ECDSA_PRIME256V1_CERT = parse_certificate(ECDSA_PRIME256V1_CERT_STR)
ECDSA_SECP384r1_CERT_STR = """
-----BEGIN CERTIFICATE-----
MIICjjCCAhMCCQD2UadeQ7ub1jAKBggqhkjOPQQDAjCBrzELMAkGA1UEBhMCVVMx
EzARBgNVBAgMCkNhbGlmb3JuaWExEjAQBgNVBAcMCUxvcyBHYXRvczEjMCEGA1UE
CgwaTGVtdXJUcnVzdCBFbnRlcnByaXNlcyBMdGQxJjAkBgNVBAsMHVVuaXR0ZXN0
aW5nIE9wZXJhdGlvbnMgQ2VudGVyMSowKAYDVQQDDCFMZW11clRydXN0IFVuaXR0
ZXN0cyBSb290IENBIDIwMTgwHhcNMTkwMjI2MTgxODU2WhcNMjkwMjIzMTgxODU2
WjCBrzELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNhbGlmb3JuaWExEjAQBgNVBAcM
CUxvcyBHYXRvczEjMCEGA1UECgwaTGVtdXJUcnVzdCBFbnRlcnByaXNlcyBMdGQx
JjAkBgNVBAsMHVVuaXR0ZXN0aW5nIE9wZXJhdGlvbnMgQ2VudGVyMSowKAYDVQQD
DCFMZW11clRydXN0IFVuaXR0ZXN0cyBSb290IENBIDIwMTgwdjAQBgcqhkjOPQIB
BgUrgQQAIgNiAARuKyHIRp2e6PB5UcY8L/bUdavkL5Zf3IegNKvaAsvkDenhDGAI
zwWgsk3rOo7jmpMibn7yJQn404uZovwyeKcApn8uVv8ltheeYAx+ySzzn/APxNGy
cye/nv1D9cDW628wCgYIKoZIzj0EAwIDaQAwZgIxANl1ljDH4ykNK2OaRqKOkBOW
cKk1SvtiEZDS/wytiZGCeaxYteSYF+3GE8V2W1geWAIxAI8D7DY0HU5zw+oxAlTD
Uw/TeHA6q0QV4otPvrINW3V09iXDwFSPe265fTkHSfT6hQ==
-----END CERTIFICATE-----
"""
ECDSA_SECP384r1_CERT = parse_certificate(ECDSA_SECP384r1_CERT_STR)
DSA_CERT_STR = """
-----BEGIN CERTIFICATE-----
MIIDmTCCA1YCCQD5h/cM7xYO9jALBglghkgBZQMEAwIwga8xCzAJBgNVBAYTAlVT
MRMwEQYDVQQIDApDYWxpZm9ybmlhMRIwEAYDVQQHDAlMb3MgR2F0b3MxIzAhBgNV
BAoMGkxlbXVyVHJ1c3QgRW50ZXJwcmlzZXMgTHRkMSYwJAYDVQQLDB1Vbml0dGVz
dGluZyBPcGVyYXRpb25zIENlbnRlcjEqMCgGA1UEAwwhTGVtdXJUcnVzdCBVbml0
dGVzdHMgUm9vdCBDQSAyMDE4MB4XDTE5MDIyNjE4MjUyMloXDTI5MDIyMzE4MjUy
Mlowga8xCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDYWxpZm9ybmlhMRIwEAYDVQQH
DAlMb3MgR2F0b3MxIzAhBgNVBAoMGkxlbXVyVHJ1c3QgRW50ZXJwcmlzZXMgTHRk
MSYwJAYDVQQLDB1Vbml0dGVzdGluZyBPcGVyYXRpb25zIENlbnRlcjEqMCgGA1UE
AwwhTGVtdXJUcnVzdCBVbml0dGVzdHMgUm9vdCBDQSAyMDE4MIIBtjCCASsGByqG
SM44BAEwggEeAoGBAO2+6wO20rn9K7RtXJ7/kCSVFzYZsY1RKvmJ6BBkMFIepBkz
2pk62tRhJgNH07GKF7pyTPRRKqt38CaPK4ERUpavx3Ok6vZ3PKq8tMac/PMKBmT1
Xfpch54KDlCdreEMJqYiCwbIyiSCR4+PCH+7xC5Uh0PIZo6otNWe3Wkk53CfAhUA
8d4YAtto6D30f7qkEa7DMAccUS8CgYAiv8r0k0aUEaeioblcCAjmhvE0v8/tD5u1
anHO4jZIIv7uOrNFIGfqcNEOBs5AQkt5Bxn6x0b/VvtZ0FSrD0j4f36pTgro6noG
/0oRt0JngxsMSfo0LV4+bY62v21A0SneNgTgY+ugdfgGWvb0+9tpsIhiY69T+7c8
Oa0S6OWSPAOBhAACgYB5wa+nJJNZPoTWFum27JlWGYLO2flg5EpWlOvcEE0o5RfB
FPnMM033kKQQEI0YpCAq9fIMKhhUMk1X4mKUBUTt+Nrn1pY2l/wt5G6AQdHI8QXz
P1ecBbHPNZtWe3iVnfOgz/Pd8tU9slcXP9z5XbZ7R/oGcF/TPRTtbLEkYZNaDDAL
BglghkgBZQMEAwIDMAAwLQIVANubSNMSLt8plN9ZV3cp4pe3lMYCAhQPLLE7rTgm
92X+hWfyz000QEpYEQ==
-----END CERTIFICATE-----
"""
DSA_CERT = parse_certificate(DSA_CERT_STR)