Orphaned certificates (#406)

* Fixing whitespace.

* Fixing syncing.

* Fixing tests
This commit is contained in:
kevgliss
2016-07-28 13:08:24 -07:00
committed by GitHub
parent a644f45625
commit 29a330b1f4
13 changed files with 199 additions and 174 deletions

View File

@ -7,6 +7,7 @@
"""
import datetime
import lemur.common.utils
from flask import current_app
from sqlalchemy.orm import relationship
@ -38,7 +39,7 @@ class Certificate(db.Model):
__tablename__ = 'certificates'
id = Column(Integer, primary_key=True)
owner = Column(String(128), nullable=False)
name = Column(String(128)) # , unique=True) TODO make all names unique
name = Column(String(128), unique=True)
description = Column(String(1024))
active = Column(Boolean, default=True)
@ -78,7 +79,7 @@ class Certificate(db.Model):
endpoints = relationship("Endpoint", backref='certificate')
def __init__(self, **kwargs):
cert = defaults.parse_certificate(kwargs['body'])
cert = lemur.common.utils.parse_certificate(kwargs['body'])
self.issuer = defaults.issuer(cert)
self.cn = defaults.common_name(cert)
@ -88,14 +89,19 @@ class Certificate(db.Model):
# when destinations are appended they require a valid name.
if kwargs.get('name'):
self.name = kwargs['name']
self.name = get_or_increase_name(kwargs['name'])
else:
self.name = get_or_increase_name(defaults.certificate_name(self.cn, self.issuer, self.not_before, self.not_after, self.san))
self.owner = kwargs['owner']
self.body = kwargs['body']
self.private_key = kwargs.get('private_key')
self.chain = kwargs.get('chain')
self.body = kwargs['body'].strip()
if kwargs.get('private_key'):
self.private_key = kwargs['private_key'].strip()
if kwargs.get('chain'):
self.chain = kwargs['chain'].strip()
self.destinations = kwargs.get('destinations', [])
self.notifications = kwargs.get('notifications', [])
self.description = kwargs.get('description')

View File

@ -77,16 +77,16 @@ def get_by_source(source_label):
return Certificate.query.filter(Certificate.sources.any(label=source_label))
def find_duplicates(cert_body):
def find_duplicates(cert):
"""
Finds certificates that already exist within Lemur. We do this by looking for
certificate bodies that are the same. This is the most reliable way to determine
if a certificate is already being tracked by Lemur.
:param cert_body:
:param cert:
:return:
"""
return Certificate.query.filter_by(body=cert_body).all()
return Certificate.query.filter_by(body=cert['body'].strip(), chain=cert['chain'].strip()).all()
def export(cert, export_plugin):
@ -172,14 +172,9 @@ def import_certificate(**kwargs):
:param kwargs:
"""
from lemur.users import service as user_service
if not kwargs.get('owner'):
kwargs['owner'] = current_app.config.get('LEMUR_SECURITY_TEAM_EMAIL')[0]
if not kwargs.get('creator'):
kwargs['creator'] = user_service.get_by_email('lemur@nobody')
return upload(**kwargs)
@ -187,7 +182,6 @@ def upload(**kwargs):
"""
Allows for pre-made certificates to be imported into Lemur.
"""
from lemur.users import service as user_service
roles = create_certificate_roles(**kwargs)
if kwargs.get('roles'):
@ -202,8 +196,7 @@ def upload(**kwargs):
try:
g.user.certificates.append(cert)
except AttributeError:
user = user_service.get_by_email('lemur@nobody')
user.certificates.append(cert)
current_app.logger.debug("No user to associate uploaded certificate to.")
return database.update(cert)