Merge branch 'master' into skip_duplicate_tasks
This commit is contained in:
@ -7,6 +7,7 @@
|
||||
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
|
||||
"""
|
||||
import random
|
||||
import re
|
||||
import string
|
||||
|
||||
import sqlalchemy
|
||||
@ -67,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.
|
||||
|
@ -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
|
||||
|
Reference in New Issue
Block a user