Reducing the stacked exceptions plus a bit of pep8
This commit is contained in:
parent
b3c4324728
commit
7dbca821c3
@ -363,7 +363,10 @@ def check_revoked():
|
|||||||
else:
|
else:
|
||||||
status = verify_string(cert.body, "")
|
status = verify_string(cert.body, "")
|
||||||
|
|
||||||
cert.status = 'valid' if status else 'revoked'
|
if status is None:
|
||||||
|
cert.status = 'unknown'
|
||||||
|
else:
|
||||||
|
cert.status = 'valid' if status else 'revoked'
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
sentry.captureException()
|
sentry.captureException()
|
||||||
|
@ -7,6 +7,7 @@
|
|||||||
"""
|
"""
|
||||||
import requests
|
import requests
|
||||||
import subprocess
|
import subprocess
|
||||||
|
from flask import current_app
|
||||||
from requests.exceptions import ConnectionError, InvalidSchema
|
from requests.exceptions import ConnectionError, InvalidSchema
|
||||||
from cryptography import x509
|
from cryptography import x509
|
||||||
from cryptography.hazmat.backends import default_backend
|
from cryptography.hazmat.backends import default_backend
|
||||||
@ -14,6 +15,7 @@ from cryptography.hazmat.backends import default_backend
|
|||||||
from lemur.utils import mktempfile
|
from lemur.utils import mktempfile
|
||||||
from lemur.common.utils import parse_certificate
|
from lemur.common.utils import parse_certificate
|
||||||
|
|
||||||
|
crl_cache = {}
|
||||||
|
|
||||||
def ocsp_verify(cert_path, issuer_chain_path):
|
def ocsp_verify(cert_path, issuer_chain_path):
|
||||||
"""
|
"""
|
||||||
@ -29,6 +31,10 @@ def ocsp_verify(cert_path, issuer_chain_path):
|
|||||||
p1 = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
p1 = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||||
url, err = p1.communicate()
|
url, err = p1.communicate()
|
||||||
|
|
||||||
|
if not url:
|
||||||
|
current_app.logger.debug("No OCSP URL in certificate")
|
||||||
|
return None
|
||||||
|
|
||||||
p2 = subprocess.Popen(['openssl', 'ocsp', '-issuer', issuer_chain_path,
|
p2 = subprocess.Popen(['openssl', 'ocsp', '-issuer', issuer_chain_path,
|
||||||
'-cert', cert_path, "-url", url.strip()], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
'-cert', cert_path, "-url", url.strip()], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||||
|
|
||||||
@ -57,28 +63,39 @@ def crl_verify(cert_path):
|
|||||||
:raise Exception: If certificate does not have CRL
|
:raise Exception: If certificate does not have CRL
|
||||||
"""
|
"""
|
||||||
with open(cert_path, 'rt') as c:
|
with open(cert_path, 'rt') as c:
|
||||||
cert = parse_certificate(c.read())
|
try:
|
||||||
|
cert = parse_certificate(c.read())
|
||||||
|
except Exception as e:
|
||||||
|
current_app.logger.error(e)
|
||||||
|
return None
|
||||||
|
|
||||||
distribution_points = cert.extensions.get_extension_for_oid(x509.OID_CRL_DISTRIBUTION_POINTS).value
|
try:
|
||||||
|
distribution_points = cert.extensions.get_extension_for_oid(
|
||||||
|
x509.OID_CRL_DISTRIBUTION_POINTS
|
||||||
|
).value
|
||||||
|
except x509.ExtensionNotFound:
|
||||||
|
current_app.logger.warn("No CRLDP extension in certificate")
|
||||||
|
return None
|
||||||
|
|
||||||
for p in distribution_points:
|
for p in distribution_points:
|
||||||
point = p.full_name[0].value
|
point = p.full_name[0].value
|
||||||
|
|
||||||
try:
|
if point not in crl_cache:
|
||||||
response = requests.get(point)
|
try:
|
||||||
|
response = requests.get(point)
|
||||||
|
|
||||||
if response.status_code != 200:
|
if response.status_code != 200:
|
||||||
|
raise Exception("Unable to retrieve CRL: {0}".format(point))
|
||||||
|
except InvalidSchema:
|
||||||
|
# Unhandled URI scheme (like ldap://); skip this distribution point.
|
||||||
|
continue
|
||||||
|
except ConnectionError:
|
||||||
raise Exception("Unable to retrieve CRL: {0}".format(point))
|
raise Exception("Unable to retrieve CRL: {0}".format(point))
|
||||||
except InvalidSchema:
|
|
||||||
# Unhandled URI scheme (like ldap://); skip this distribution point.
|
|
||||||
continue
|
|
||||||
except ConnectionError:
|
|
||||||
raise Exception("Unable to retrieve CRL: {0}".format(point))
|
|
||||||
|
|
||||||
crl = x509.load_der_x509_crl(response.content, backend=default_backend())
|
crl_cache[point] = x509.load_der_x509_crl(response.content, backend=default_backend())
|
||||||
|
|
||||||
for r in crl:
|
for r in crl_cache[point]:
|
||||||
if cert.serial == r.serial_number:
|
if cert.serial_number == r.serial_number:
|
||||||
try:
|
try:
|
||||||
reason = r.extensions.get_extension_for_class(x509.CRLReason).value
|
reason = r.extensions.get_extension_for_class(x509.CRLReason).value
|
||||||
# Handle "removeFromCRL" revoke reason as unrevoked; continue with the next distribution point.
|
# Handle "removeFromCRL" revoke reason as unrevoked; continue with the next distribution point.
|
||||||
@ -86,6 +103,7 @@ def crl_verify(cert_path):
|
|||||||
if reason == x509.ReasonFlags.remove_from_crl:
|
if reason == x509.ReasonFlags.remove_from_crl:
|
||||||
break
|
break
|
||||||
except x509.ExtensionNotFound:
|
except x509.ExtensionNotFound:
|
||||||
|
current_app.logger.warn("extension not found in CRL?")
|
||||||
pass
|
pass
|
||||||
|
|
||||||
return
|
return
|
||||||
@ -103,13 +121,15 @@ def verify(cert_path, issuer_chain_path):
|
|||||||
"""
|
"""
|
||||||
# OCSP is our main source of truth, in a lot of cases CRLs
|
# OCSP is our main source of truth, in a lot of cases CRLs
|
||||||
# have been deprecated and are no longer updated
|
# have been deprecated and are no longer updated
|
||||||
try:
|
verify_result = ocsp_verify(cert_path, issuer_chain_path)
|
||||||
return ocsp_verify(cert_path, issuer_chain_path)
|
|
||||||
except Exception as e:
|
if verify_result is None:
|
||||||
try:
|
verify_result = crl_verify(cert_path)
|
||||||
return crl_verify(cert_path)
|
|
||||||
except Exception as e:
|
if verify_result is None:
|
||||||
raise Exception("Failed to verify")
|
current_app.logger.warn("Failed to verify")
|
||||||
|
|
||||||
|
return verify_result
|
||||||
|
|
||||||
|
|
||||||
def verify_string(cert_string, issuer_string):
|
def verify_string(cert_string, issuer_string):
|
||||||
|
Loading…
Reference in New Issue
Block a user