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

Compare certificate hashes to determine if Lemur already has a synced certificate

This commit is contained in:
Curtis Castrapel 2019-01-14 13:35:55 -08:00
parent 2c545aa4bc
commit 3567a768d5
2 changed files with 14 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
lemur/common/utils.py
View File

@ -12,6 +12,7 @@ import string
import sqlalchemy import sqlalchemy
from cryptography import x509 from cryptography import x509
from cryptography.hazmat.backends import default_backend 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
from cryptography.hazmat.primitives.serialization import load_pem_private_key from cryptography.hazmat.primitives.serialization import load_pem_private_key
from flask_restful.reqparse import RequestParser from flask_restful.reqparse import RequestParser
@ -226,3 +227,13 @@ def truthiness(s):
"""If input string resembles something truthy then return True, else False.""" """If input string resembles something truthy then return True, else False."""
return s.lower() in ('true', 'yes', 'on', 't', '1') return s.lower() in ('true', 'yes', 'on', 't', '1')
def find_matching_certificates_by_hash(cert, matching_certs):
"""Given a Cryptography-formatted certificate cert, and Lemur-formatted certificates (matching_certs),
determine if any of the certificate hashes match and return the matches."""
matching = []
for c in matching_certs:
if parse_certificate(c.body).fingerprint(hashes.SHA256()) == cert.fingerprint(hashes.SHA256()):
matching.append(c)
return matching

5
lemur/sources/service.py
View File

@ -17,7 +17,7 @@ from lemur.endpoints import service as endpoint_service
from lemur.destinations import service as destination_service from lemur.destinations import service as destination_service
from lemur.certificates.schemas import CertificateUploadInputSchema from lemur.certificates.schemas import CertificateUploadInputSchema
from lemur.common.utils import parse_certificate from lemur.common.utils import find_matching_certificates_by_hash, parse_certificate
from lemur.common.defaults import serial from lemur.common.defaults import serial
from lemur.plugins.base import plugins from lemur.plugins.base import plugins
@ -126,7 +126,8 @@ def sync_certificates(source, user):
if not exists: if not exists:
cert = parse_certificate(certificate['body']) cert = parse_certificate(certificate['body'])
exists = certificate_service.get_by_serial(serial(cert)) matching_serials = certificate_service.get_by_serial(serial(cert))
exists = find_matching_certificates_by_hash(cert, matching_serials)
if not certificate.get('owner'): if not certificate.get('owner'):
certificate['owner'] = user.email certificate['owner'] = user.email