Merge pull request #14 from gattjoe/1.8

v1.8

CHANGELOG.md

@@ -29,3 +29,10 @@
 - Upgraded cryptography to 3.4.6
 - Fixed failing tests
 - Added tox tests for Python 3.9
+
+# v1.8.0
+- Fixed a bug to handle a situation when parsing a certificate with an AIA extension but no OCSP URL
+- Fixed a bug to handle a situation where the remote host is not using SSL/TLS and we attempt to do a SSL/TLS handshake
+- Fixed a bug to handle a situation where the remote host does not respond to a Client Hello
+- Prepended all exceptions with the function name for easier troubleshooting
+- Upgraded cryptography to 3.4.7 to support the latest versions of OpenSSL

README.md

@@ -5,31 +5,26 @@
 [![Python version](https://img.shields.io/pypi/pyversions/ocsp-checker.svg)](https://pypi.org/project/ocsp-checker/)
 
 ## Overview
------------
 
 OCSP-Checker is a python package based on Alban Diquet's [nassl](https://github.com/nabla-c0d3/nassl) wrapper and the Python Cryptographic Authority's [cryptography](https://github.com/pyca/cryptography) package. Relying on a web browser to check the revocation status of a x509 digital certificate [has](https://www.imperialviolet.org/2014/04/19/revchecking.html) [been](https://www.imperialviolet.org/2014/04/29/revocationagain.html) [broken](https://scotthelme.co.uk/revocation-is-broken/) from the beginning, and validating certificates outside of the web browser is a manual process. OCSP-Checker aims to solve this by providing an automated means to check the [OCSP](https://en.wikipedia.org/wiki/Online_Certificate_Status_Protocol) revocation status for a x509 digital certificate.
 
 
 ## Pre-requisites
------------------
 
 __Python__ - Python 3.7 (64-bit) and above.
 
 ## Installation
----------------
 
 ```pip install ocsp-checker```
 
 ## Usage
---------
 
 ```
 >>> from ocspchecker import ocspchecker
 >>> ocsp_request = ocspchecker.get_ocsp_status("github.com")
 ```
 
 ## Sample Output
-----------------
 
 Sample output below, let me know if you want to add more fields/information to the output.
 
@@ -43,7 +38,6 @@ PLEASE NOTE: If you run this on a network with a MITM SSL proxy, you may receive
 ```
 
 ## Command Line Usage
----------------------
 
 OCSP-Checker can now be used at the command line. The format is:
 ```

ocspchecker/__init__.py

@@ -1,7 +1,7 @@
 # __init__.py
 
 __title__ = "ocsp-checker"
-__version__ = "1.7.0"
+__version__ = "1.8.0"
 __author__ = "Joe Gatt"
 
 from ocspchecker.ocspchecker import get_ocsp_status

ocspchecker/ocspchecker.py

@@ -23,7 +23,7 @@ def get_ocsp_status(host: str, port: Any = None) -> list:
     """ Main function with two inputs: host, and port.
     Port defaults to TCP 443 """
 
-    results = []
+    results: list = []
     results.append(f"Host: {host}:{port}")
 
     # pylint: disable=W0703
@@ -49,34 +49,19 @@ def get_ocsp_status(host: str, port: Any = None) -> list:
         return results
 
     try:
+        # Get the remote certificate chain
         cert_chain = get_certificate_chain(host, port)
 
-    except Exception as err:
-        results.append("Error: " + str(err))
-        return results
-
-    try:
+        # Extract OCSP URL from leaf certificate
         ocsp_url = extract_ocsp_url(cert_chain)
 
-    except Exception as err:
-        results.append("Error: " + str(err))
-        return results
-
-    try:
+        # Build OCSP request
         ocsp_request = build_ocsp_request(cert_chain)
 
-    except Exception as err:
-        results.append("Error: " + str(err))
-        return results
-
-    try:
+        # Send OCSP request to responder and get result
         ocsp_response = get_ocsp_response(ocsp_url, ocsp_request)
 
-    except Exception as err:
-        results.append("Error: " + str(err))
-        return results
-
-    try:
+        # Extract OCSP result from OCSP response
         ocsp_result = extract_ocsp_result(ocsp_response)
 
     except Exception as err:
@@ -91,10 +76,11 @@ def get_ocsp_status(host: str, port: Any = None) -> list:
 
 def get_certificate_chain(host: str, port: int) -> List[str]:
 
-    """ Connect to the host on the port and obtain certificate chain.
-     TODO: Tests against WantReadError and WantX509LookupError needed. """
+    """ Connect to the host on the port and obtain certificate chain  """
+
+    func_name: str = "get_certificate_chain"
 
-    cert_chain = []
+    cert_chain: list = []
 
     soc = socket(AF_INET, SOCK_STREAM, proto=0)
     soc.settimeout(3)
@@ -103,42 +89,49 @@ def get_certificate_chain(host: str, port: int) -> List[str]:
         soc.connect((host, port))
 
     except gaierror:
-        raise Exception(f"{host}:{port} is invalid or not known.") from None
+        raise Exception(f"{func_name}: {host}:{port} is invalid or not known.") from None
 
     except timeout:
-        raise Exception(f"Connection to {host}:{port} timed out.") from None
+        raise Exception(f"{func_name}: Connection to {host}:{port} timed out.") from None
 
-    except OverflowError:
-        raise Exception(f"Illegal port: {port}. Port must be between 0-65535.") from None
+    except (OverflowError, TypeError):
+        raise Exception(f"{func_name}: Illegal port: {port}. Port must be between 0-65535.") from None
 
-    except TypeError:
-        raise Exception(f"Illegal port: {port}. Port must be between 0-65535.") from None
+    except ConnectionRefusedError:
+        raise Exception(f"{func_name}: Connection to {host}:{port} refused.") from None
 
     ssl_client = SslClient(
         ssl_version=OpenSslVersionEnum.SSLV23,
         underlying_socket=soc,
         ssl_verify=OpenSslVerifyEnum.NONE
     )
 
-    # Add Server Name Indication (SNI) extension to the CLIENT HELLO
+    # Add Server Name Indication (SNI) extension to the Client Hello
     ssl_client.set_tlsext_host_name(host)
 
     try:
         ssl_client.do_handshake()
         cert_chain = ssl_client.get_received_chain()
 
+    except IOError as err:
+        raise ValueError(f"{func_name}: {host} did not respond to the Client Hello.") from None
+
     except WantReadError as err:
-        raise ValueError(err.strerror) from None
+        raise ValueError(f"{func_name}: err.strerror") from None
 
     except WantX509LookupError as err:
-        raise ValueError(err.strerror) from None
+        raise ValueError(f"{func_name}: err.strerror") from None
 
     except OpenSSLError as err:
-        raise ValueError(err) from None
+        if "1408F10B" in err.args[0]:
+            # https://github.com/openssl/openssl/issues/6805
+            raise ValueError(f"{func_name}: Remote host is not using SSL/TLS on port: {port}") from None
+
+        raise ValueError(f"{func_name}: err") from None
 
     finally:
+        # shutdown() will also close the underlying socket
         ssl_client.shutdown()
-        soc = None
 
     return cert_chain
 
@@ -149,7 +142,9 @@ def extract_ocsp_url(cert_chain: List[str]) -> str:
      access location AUTHORITY_INFORMATION_ACCESS extensions to
      get the ocsp url """
 
-    ocsp_url = ""
+    func_name: str = "extract_ocsp_url"
+
+    ocsp_url: str = ""
 
     # Convert to a certificate object in cryptography.io
     certificate = x509.load_pem_x509_certificate(
@@ -167,9 +162,12 @@ def extract_ocsp_url(cert_chain: List[str]) -> str:
             if aia_extensions[index].access_method._name == "OCSP":
                 ocsp_url = aia_extensions[index].access_location.value
 
+        if ocsp_url == "":
+            raise ValueError(f"{func_name}: OCSP URL missing from Certificate AIA Extension.")
+
     except ExtensionNotFound:
         raise ValueError(
-            "Certificate Authority Information Access (AIA) Extension Missing. Possible MITM Proxy."
+            f"{func_name}: Certificate Authority Information Access (AIA) Extension Missing. Possible MITM Proxy."
         ) from None
 
     return ocsp_url
@@ -181,6 +179,8 @@ def build_ocsp_request(cert_chain: List[str]) -> bytes:
      see: https://cryptography.io/en/latest/x509/ocsp/#cryptography.x509.ocsp.OCSPRequestBuilder
      for more information """
 
+    func_name: str = "build_ocsp_request"
+
     try:
         leaf_cert = x509.load_pem_x509_certificate(
             str.encode(cert_chain[0]), default_backend()
@@ -190,7 +190,7 @@ def build_ocsp_request(cert_chain: List[str]) -> bytes:
         )
 
     except ValueError:
-        raise Exception("Unable to load x509 certificate.") from None
+        raise Exception(f"{func_name}: Unable to load x509 certificate.") from None
 
     # Build OCSP request
     builder = ocsp.OCSPRequestBuilder()
@@ -205,9 +205,11 @@ def get_ocsp_response(ocsp_url: str, ocsp_request_data: bytes):
 
     """ Send OCSP request to ocsp responder and retrieve response """
 
+    func_name: str = "get_ocsp_response"
+
     # Confirm that the ocsp_url is a valid url
     if not url(ocsp_url):
-        raise Exception(f"URL failed validation for {ocsp_url}")
+        raise Exception(f"{func_name}: URL failed validation for {ocsp_url}")
 
     try:
         ocsp_response = requests.post(
@@ -218,13 +220,13 @@ def get_ocsp_response(ocsp_url: str, ocsp_request_data: bytes):
         )
 
     except requests.exceptions.Timeout:
-        raise Exception(f"Request timeout for {ocsp_url}") from None
+        raise Exception(f"{func_name}: Request timeout for {ocsp_url}") from None
 
     except requests.exceptions.ConnectionError:
-        raise Exception(f"Unknown Connection Error to {ocsp_url}") from None
+        raise Exception(f"{func_name}: Unknown Connection Error to {ocsp_url}") from None
 
     except requests.exceptions.RequestException:
-        raise Exception(f"Unknown Connection Error to {ocsp_url}") from None
+        raise Exception(f"{func_name}: Unknown Connection Error to {ocsp_url}") from None
 
     return ocsp_response
 
@@ -233,6 +235,8 @@ def extract_ocsp_result(ocsp_response):
 
     """ Extract the OCSP result from the provided ocsp_response """
 
+    func_name: str = "extract_ocsp_result"
+
     try:
         ocsp_response = ocsp.load_der_ocsp_response(ocsp_response.content)
         # OCSP Response Status here:
@@ -248,14 +252,14 @@ def extract_ocsp_result(ocsp_response):
             # UNAUTHORIZED = 6
             ocsp_response = str(ocsp_response.response_status)
             ocsp_response = ocsp_response.split(".")
-            raise Exception(f"OCSP Request Error: {ocsp_response[1]}")
+            raise Exception(f"{func_name}: OCSP Request Error: {ocsp_response[1]}")
 
         certificate_status = str(ocsp_response.certificate_status)
         certificate_status = certificate_status.split(".")
         return f"OCSP Status: {certificate_status[1]}"
 
     except ValueError as err:
-        return f"{str(err)}"
+        return f"{func_name}: {str(err)}"
 
 
 def verify_port(port: Any) -> int:

requirements.txt

@@ -1,4 +1,4 @@
-cryptography==3.4.6
+cryptography==3.4.7
 nassl==4.0.0
 requests>=2.24.0
 validators==0.18.2

setup.py

@@ -8,7 +8,7 @@
 
 setup(
     name="ocsp-checker",
-    version="1.7.0",
+    version="1.8.0",
     description="Library used to check the OCSP revocation status for a x509 digital certificate.",
     long_description=long_description,
     long_description_content_type="text/markdown",
@@ -34,7 +34,7 @@
     entry_points={"console_scripts": ["ocspchecker = ocspchecker.__main__:main"]},
     # Dependencies
     install_requires=[
-        "cryptography==3.4.6",
+        "cryptography==3.4.7",
         "nassl==4.0.0",
         "requests>=2.24",
         "validators>=0.18"