Merge pull request #7 from astrobl1904/develop

Develop

Author: astrobl1904

Custom Sensors/python/starttls_certificate_sensor.py

@@ -3,16 +3,17 @@
 Monitor certificates of services that require STARTTLS and return a JSON formatted sensor result.
 
 This custom Python script sensor is used to monitor certificates of services that require STARTTLS
-to initiate a secure transport. It takes the same parameter as the PRTG built-in sensor `SSL Certificate`
-but additionally requires the protocol the sensor must use to communicate with the remote endpoint.
+to initiate a secure transport. It takes the same parameter as the PRTG built-in sensor
+`SSL Certificate` but additionally requires the protocol the sensor must use to communicate with
+the remote endpoint.
 The list of protocols is currently limited to `SMTP`, `LMTP`, and `LDAP`.
 
 The sensor result in JSON contains the same channels as the `SSL Certificate` sensor with channel
 `Days to Expiration` set as primary channel.
 
 Keyword for additional parameters:
 port                        -- Port for the connection to the target endpoint (default: 25)
-protocol                    -- Protocol used for the connection to the target endpoint (default: smtp)
+protocol                    -- Protocol used with the connection (default: smtp)
                                Implemented protocols: smtp, lmtp, ldap
 cert_domainname             -- Common Name as contained in the certificate
 cert_domainname_validation  -- Type of validation of the cert domain name (default: None)
@@ -33,47 +34,53 @@
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes
 
-
-from prtg.sensor.result import CustomSensorResult 
+from prtg.sensor.result import CustomSensorResult
 from prtg.sensor.units import ValueUnit
 
 class Protocol(Enum):
+    """Application Layer protocols
+    """
     SMTP = 1
     LMTP = 2
     IMAP = 3
     LDAP = 4
 
 class Validation(Enum):
+    """Certificate Name validations
+    """
     NONE = 1
     CN = 2
     CN_SAN = 3
 
 def prtg_params_dict(params: str) -> dict:
     """
     prtg_params_dict - Converts a PRTG params string into a dictionary.
-    
+
     It takes the params string and converts it via json into a dictionary. The solution is based
     on Stack Overflow (https://stackoverflow.com/questions/47663809/python-convert-string-to-dict)
     """
-    
+
     _params = '{' + params.strip() + '}'
     _params_json_string = ''
 
     # Remove surrounding spaces arround separator chars
-    _params_stripped = re.sub(r'\s*([:,])\s*', '\g<1>', _params)
+    _params_stripped = re.sub(r'\s*([:,])\s*', r'\g<1>', _params)
     _params_json_string = re.sub(r'([:,])', r'"\g<1>"', _params_stripped)
     _params_json_string = re.sub(r'{', r'{"', _params_json_string)
     _params_json_string = re.sub(r'}', r'"}', _params_json_string)
-    
+
     return json.loads(_params_json_string)
 
-def starttls_getpeercert(host: str, port: int, starttls_proto: Protocol, cert_hostname=None, timeout=3.0, msglen=4096) -> dict:
+def starttls_getpeercert(address,
+                         starttls_proto: Protocol,
+                         cert_hostname=None,
+                         timeout=3.0, msglen=4096) -> dict:
     """
     starttls_getpeercert - Retrieves the certificate of the other side of the connection.
 
     @Returns: Certificate dict with selfSigned? and rootAuthorityTrusted? mixed in.
 
-    @NOTE: Dictionary keys
+    @NOTE: Dictionary keys:
         subject (string; distinguished name form)
         issuer (string; distinguished name form)
         version (int)
@@ -91,23 +98,28 @@ def starttls_getpeercert(host: str, port: int, starttls_proto: Protocol, cert_ho
         rootAuthorityTrusted? (boolean)
 
     Ref: https://stackoverflow.com/questions/5108681/use-python-to-get-an-smtp-server-certificate
-    REf: https://stackoverflow.com/questions/71114085/how-can-i-retrieve-openldap-servers-starttls-certificate-with-pythons-ssl-libr
+    REf: https://stackoverflow.com/questions/71114085/how-can-i-retrieve-openldap-servers-starttls-
+         certificate-with-pythons-ssl-libr
     """
 
-    if cert_hostname == None:
-        sni_hostname = host
+    if cert_hostname is None:
+        sni_hostname = address[0]
     else:
         sni_hostname = cert_hostname
-    
-    # Request #1: Get certificate data (incl. self-issued certs) with context in verify_mode == CERT_NONE
-    # The SSLContext is created with the class constructor since retrieving self-signed certs require
-    # loosened SSL settings. To fetch self-signed certs verify_mode MUST be set to CERT_NONE which
-    # requires disabling hostname checking.
+
+    # Request #1: Get certificate data (incl. self-issued certs) with verify_mode set to CERT_NONE
+    # in the created context
+    # The SSLContext is created with the class constructor since retrieving self-signed certs
+    # require loosened SSL settings. To fetch self-signed certs verify_mode MUST be set to
+    # CERT_NONE which requires disabling hostname checking.
     ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
     ctx.check_hostname = False
-    ctx.verify_mode = ssl.VerifyMode.CERT_NONE
+    ctx.verify_mode = ssl.CERT_NONE
 
-    raw_sock = _starttls_do_service_handshake((host, port), starttls_proto, timeout=timeout, msglen=msglen)
+    raw_sock = _starttls_do_service_handshake(address,
+                                              starttls_proto,
+                                              timeout=timeout,
+                                              msglen=msglen)
     with ctx.wrap_socket(raw_sock, server_hostname=sni_hostname) as ssl_sock:
         cert_der = ssl_sock.getpeercert(binary_form=True)
         cert = x509.load_der_x509_certificate(cert_der, default_backend())
@@ -119,27 +131,33 @@ def starttls_getpeercert(host: str, port: int, starttls_proto: Protocol, cert_ho
     if cert_dict['selfSigned?']:
         cert_dict['rootAuthorityTrusted?'] = False
     else:
-        ctx_trust_check = ssl.create_default_context()
-        raw_sock_trust_check = _starttls_do_service_handshake((host, port), starttls_proto, timeout=timeout, msglen=msglen)
+        ctx = ssl.create_default_context()
+        raw_sock = _starttls_do_service_handshake(address,
+                                                  starttls_proto,
+                                                  timeout=timeout,
+                                                  msglen=msglen)
         try:
-            ssl_sock_trust_check = ctx_trust_check.wrap_socket(raw_sock_trust_check, server_hostname=sni_hostname)
+            ssl_sock = ctx.wrap_socket(raw_sock, server_hostname=sni_hostname)
             cert_dict['rootAuthorityTrusted?'] = True
-            ssl_sock_trust_check.close()
+            ssl_sock.close()
         except ssl.SSLCertVerificationError:
             cert_dict['rootAuthorityTrusted?'] = False
 
     return cert_dict
 
-def _starttls_do_service_handshake(address, starttls_proto: Protocol, timeout=3.0, msglen=4096) -> socket.socket:
+def _starttls_do_service_handshake(address,
+                                   starttls_proto: Protocol,
+                                   timeout=3.0,
+                                   msglen=4096) -> socket.socket:
     """
     starttls_do_service_handshake - Perform the service handshake to initiate a TLS connection
 
     @Returns: socket.socket
     """
     # Protocol.LDAP sends a LDAP_START_TLS_OID - sniffed with Wireshark
     protocol_greeters = {
-        Protocol.SMTP: bytes("EHLO {0}\nSTARTTLS\n".format(socket.gethostname()), 'ascii'),
-        Protocol.LMTP: bytes("LHLO {0}\nSTARTTLS\n".format(socket.gethostname()), 'ascii'),
+        Protocol.SMTP: bytes(f'EHLO {socket.gethostname()}\nSTARTTLS\n', 'ascii'),
+        Protocol.LMTP: bytes(f'LHLO {socket.gethostname()}\nSTARTTLS\n', 'ascii'),
         Protocol.LDAP: b'\x30\x1d\x02\x01\x01\x77\x18\x80\x16\x31\x2e\x33\x2e\x36\x2e\x31' \
                        b'\x2e\x34\x2e\x31\x2e\x31\x34\x36\x36\x2e\x32\x30\x30\x33\x37'
     }
@@ -164,14 +182,17 @@ def _starttls_do_service_handshake(address, starttls_proto: Protocol, timeout=3.
         raw_sock.send(protocol_greeting)
         # Look for \x0a\x01 {result code} \x04\x00\x04\x00 - if the second \x04 is not followed
         # by \x00 then it seems that the server support STARTTLS but has no cert installed
-        _ldap_tls_ext_response = raw_sock.recv(msglen)
-        _ldap_tls_ext_response_result_pos = _ldap_tls_ext_response.find(b'\x0a\x01')
-        _ldap_tls_ext_response_result_trail = _ldap_tls_ext_response.find(b'\x04\x00\x04\x00', _ldap_tls_ext_response_result_pos)
+        _ldap_response = raw_sock.recv(msglen)
+        _ldap_response_result_pos = _ldap_response.find(b'\x0a\x01')
+        _ldap_response_result_trail = _ldap_response.find(b'\x04\x00\x04\x00',
+                                                          _ldap_response_result_pos)
 
-        if not(_ldap_tls_ext_response_result_trail - _ldap_tls_ext_response_result_pos == 3 and
-                _ldap_tls_ext_response[_ldap_tls_ext_response_result_pos + 2] == 0):
+        if not(_ldap_response_result_trail - _ldap_response_result_pos == 3 and
+                _ldap_response[_ldap_response_result_pos + 2] == 0):
             raw_sock.close()
-            raise OSError("LDAP Server does not support LDAP_START_TLS_OID or has no certificate installed.")
+            _err_msg = "LDAP Server does not support LDAP_START_TLS_OID "
+            _err_msg += "or has no certificate installed."
+            raise OSError(_err_msg)
 
     return raw_sock
 
@@ -207,12 +228,13 @@ def _starttls_cert_dict(cert) -> dict:
         try:
             _extension = cert.extensions.get_extension_for_oid(_extension_oid)
             if _dict_key == 'subjectAltName':
-                cert_dict[_dict_key] = [ ('DNS', _dnsname) for _dnsname in _extension.value.get_values_for_type(x509.DNSName) ]
+                _extension_subject_alt_names = _extension.value.get_values_for_type(x509.DNSName)
+                cert_dict[_dict_key] = [ ('DNS', _name) for _name in _extension_subject_alt_names ]
             if _dict_key == 'crlDistributionPoints':
                 cert_dict[_dict_key] = [ _crldp.full_name[0].value for _crldp in _extension.value ]
         except x509.ExtensionNotFound:
             pass
-    
+
     # Cert extension data - Authority Access Info Methods (OCSP, caIssuers)
     _extension_oid = x509.ObjectIdentifier('1.3.6.1.5.5.7.1.1')
     _authority_access_method_oids = [
@@ -224,23 +246,24 @@ def _starttls_cert_dict(cert) -> dict:
         while len(_authority_access_method_oids) > 0:
             _access_method_oid, _dict_key = _authority_access_method_oids.pop(0)
             for _access_description in _extension.value:
-                try:
-                    if _access_description.access_method == _access_method_oid:
-                        cert_dict[_dict_key] = _access_description.access_location.value
-                        break
-                except:
-                    pass
+                if _access_description.access_method == _access_method_oid:
+                    cert_dict[_dict_key] = _access_description.access_location.value
+                    break
     except x509.ExtensionNotFound:
         pass
 
     return cert_dict
 
-def _prtg_cert_cncheck_result(certificate: dict, validation_mode: Validation, cert_hostname: str) -> int:
+def _prtg_cert_cncheck_result(certificate: dict,
+                              validation_mode: Validation,
+                              cert_hostname: str) -> int:
     """
-    prtg_cert_cncheck_result - Returns the proper result value based on the sensor parameter cert_domainname_validation
+    prtg_cert_cncheck_result - Returns the proper result value based on cert_domainname_validation
 
-    The return value matches the expected result specified in the default overlay prtg.standardlookups.sslcertificatesensor.cncheck.
-    This script DOES NOT return all values since SNI check and common_name check are considered interchangeable.
+    The return value matches the expected result specified in the default overlay
+    prtg.standardlookups.sslcertificatesensor.cncheck.
+    This script DOES NOT return all values since SNI check and common_name check
+    are considered interchangeable.
 
     Expected result values defined in overlay:
         Value 0: State Ok, Matches device address (Validation mode: cn)
@@ -262,10 +285,10 @@ def _prtg_cert_cncheck_result(certificate: dict, validation_mode: Validation, ce
         if cert_hostname.strip().lower() == certificate["commonName"].strip().lower():
             result_value = 5
         if "subjectAltName" in certificate.keys():
-            _dns_names = [altname_tuple[1] for altname_tuple in certificate["subjectAltName"] if altname_tuple[0] == "DNS"]
-            if cert_hostname.strip().lower() in _dns_names:
+            _names = [_tuple[1] for _tuple in certificate["subjectAltName"] if _tuple[0] == "DNS"]
+            if cert_hostname.strip().lower() in _names:
                 result_value = 5
-        # If after all checks result_value is still 2 (disabled) - correct it to the propper check error value
+        # If after all checks result_value is still 2 (disabled) - correct it to the error value
         if result_value == 2:
             result_value = 6
 
@@ -280,17 +303,23 @@ def main():
     in order to start a secure connection.
     """
     try:
+        # Basic argument check
+        if len(sys.argv) < 2:
+            raise TypeError("JSON object argument missing.")
         data = json.loads(sys.argv[1])
+
+        if "params" not in data.keys():
+            raise TypeError("Key 'params' missing in JSON object.")
         params = prtg_params_dict(data["params"])
-        _now = datetime.datetime.now()
 
-        cert = starttls_getpeercert(
-            data["host"],
-            int(params.get("port", "25")),
-            Protocol[params.get("protocol", "smtp").upper()],
-            cert_hostname=params.get("cert_domainname", data["host"]))
+        _now = datetime.datetime.now()
+        cert = starttls_getpeercert((data["host"], int(params.get("port", "25"))),
+                                    Protocol[params.get("protocol", "smtp").upper()],
+                                    cert_hostname=params.get("cert_domainname", data["host"]))
 
-        csr_text_ok = "OK. Certificate Common Name: {} - Certificate Thumbprint: {} - STARTTLS Protocol: {}"
+        csr_text_ok = "OK. Certificate Common Name: {} - "
+        csr_text_ok += "Certificate Thumbprint: {} - "
+        csr_text_ok += "STARTTLS Protocol: {}"
         csr = CustomSensorResult(text=csr_text_ok.format(cert['commonName'],
                                                          cert['fingerprint'],
                                                          params.get('protocol', "smtp").upper()))
@@ -307,9 +336,9 @@ def main():
                         limit_warning_msg="Certificate will expire soon.")
 
         # Channel _Common Name Check_
-        _prtg_cncheck_value = _prtg_cert_cncheck_result(cert,
-                                                            Validation[params.get("cert_domainname_validation", "NONE").upper()],
-                                                            params.get("cert_domainname", data["host"]))
+        _validation = Validation[params.get("cert_domainname_validation", "NONE").upper()]
+        _sni_domainname = params.get("cert_domainname", data["host"])
+        _prtg_cncheck_value = _prtg_cert_cncheck_result(cert, _validation, _sni_domainname)
         csr.add_channel(name="Common Name Check",
                         unit=ValueUnit.CUSTOM,
                         value_lookup='prtg.standardlookups.sslcertificatesensor.cncheck',
@@ -350,12 +379,21 @@ def main():
                 is_float=False,
                 is_limit_mode=False)
 
-        # Print sensor JSON result
-        print(csr.json_result)
-
-    except Exception as e:
+    except json.JSONDecodeError as sensor_error:
+        csr = CustomSensorResult(text="Python Script execution error")
+        csr.error = f"Failed to decode 'params' into valid JSON object: {str(sensor_error)}"
+    except KeyError as sensor_error:
         csr = CustomSensorResult(text="Python Script execution error")
-        csr.error = "Python Script execution error: {}".format(str(e))
+        csr.error = f"Invalid key in 'protocol'|'cert_domainname_validation': {str(sensor_error)}"
+    except ssl.SSLEOFError as sensor_error:
+        csr = CustomSensorResult(text="Python Script execution error")
+        csr.error = f"Protocol handshake prior to STARTTLS possibly failed': {str(sensor_error)}"
+    except (TypeError, OSError, RuntimeError) as sensor_error:
+        csr = CustomSensorResult(text="Python Script execution error")
+        csr.error = f"Python Script runtime error: {str(sensor_error)}"
+
+    finally:
+        # Print sensor JSON result
         print(csr.json_result)
 
 if __name__ == "__main__":

README.md

@@ -5,7 +5,7 @@ This repository contains a PRTG Python Script Advanced sensor to monitor SSL Cer
 ## Sensor Summary
 
     Script Language: Python 3.7+
-    Version: 0.1.0
+    Version: 1.0.0
     Author: Andreas Strobl <astroblx@asgraphics.at>
     Verified PRTG Version: 22.2.76.1705
     Dependencies: cryptography >=37.0.0, prtg == 1.0.0
@@ -14,7 +14,7 @@ This repository contains a PRTG Python Script Advanced sensor to monitor SSL Cer
 
 This custom _Python Script Advanced_ sensor will monitor SSL certificates that require a protocol handshake prior to reading certificate data, and exposes the collected data in channels similar to PRTG's built-in _SSL Certificate_ sensor.
 
-As of version v0.1.0 this sensor supports the following application layer protocols:
+As of version v1.0.0 this sensor supports the following application layer protocols:
 
 * `SMTP`: Simple Mail Transfer Protocol, [RFC 5321](https://www.rfc-editor.org/rfc/rfc5321)
 * `LMTP`: Local Mail Transfer Protocol, [RFC 2033](https://datatracker.ietf.org/doc/html/rfc2033)