C3L-NOC/tls-expiry-tracker Changeset - c11dd0c93877 · Chaos Computer Club Lëtzebuerg Projects

Changeset - c11dd0c93877

Parent rev.

Child rev.

[Not reviewed]

0 4 1

x - 11 months ago 2024-05-09 19:56:06
xbr@c3l.lu

feat: generalize the verificator for a generic abstract handler

5 files changed with 82 insertions and 84 deletions:

backend/check_domains.py

backend/generic_handler.py

backend/mail.py

backend/tls_utils.py

backend/web.py

0 comments (0 inline, 0 general)

backend/check_domains.py

➞

Show inline comments

@@ @@ -3,9 +3,7 @@ import json @@
 import ssl
 import os
 from rich.console import Console
 from web import SSLVerificator
 from mail import MailVerificator
 from generic_handler import Verificator
 if __name__ == "__main__":
     console = Console()
@@ @@ -19,16 +17,15 @@ if __name__ == "__main__": @@
     console.log("[white]Checking web domains...")
-    ssl = SSLVerificator(context)
+    v = Verificator(context)
     for web_domain in input["domains"]["web"]:
-        result = ssl.connect(web_domain, 443)
+        result = v.connect(web_domain, 443, "ssl")
         result.print(console)
     mail = MailVerificator(context)
     for smtp_entry in input["domains"]["smtp"]:
-        result = mail.connect(smtp_entry["host"], smtp_entry["port"], "smtp")
+        result = v.connect(smtp_entry["host"], smtp_entry["port"], "smtp")
         result.print(console)
     for imap_entry in input["domains"]["imap"]:
-        result = mail.connect(imap_entry["host"], imap_entry["port"], "imap")
+        result = v.connect(imap_entry["host"], imap_entry["port"], "imap")
         result.print(console)

backend/generic_handler.py

➞

Show inline comments

@@ new file 100644 @@
 from abc import ABC, abstractmethod
 import ssl
 from tls_utils import TLSDetails, EXPIRED, REVOKED, SELF_SIGNED, ROOT_NOT_TRUSTED
 class GenericHandler(ABC):
     def __init__(self, host: str, port: int, context: ssl.SSLContext):
         self.host = host
         self.port = port
         self.context = context
     @abstractmethod
     def connect(self, verification: bool) -> int:
         raise NotImplementedError()
     @staticmethod
     def create_handler(protocol: str):
         import web, mail
         if protocol == "smtp":
             return mail.SMTPHandler
         elif protocol == "imap":
             return mail.IMAPHandler
         elif protocol == "ssl" or protocol == "tls" or protocol == "https":
             return web.SSLHandler
         else:
             raise ValueError("Invalid protocol")
 class Verificator:
     def __init__(self, context: ssl.SSLContext):
         self.context = context
     def connect(self, domain: str, port: int, protocol: str) -> TLSDetails:
         handler = GenericHandler.create_handler(protocol)(domain, port, self.context)
         try:
             expiry = handler.connect(True)
             return TLSDetails(domain_name=domain, expires_in_days=expiry)
         except ssl.SSLCertVerificationError as e:
             if e.verify_code == EXPIRED:
                 expiry = handler.connect(False)
                 return TLSDetails(domain_name=domain, expires_in_days=expiry)
             elif e.verify_code == REVOKED:
                 return TLSDetails(domain_name=domain, error_message="was revoked.")
             elif e.verify_code == SELF_SIGNED:
                 return TLSDetails(domain_name=domain, error_message="is self-signed.")
             elif e.verify_code == ROOT_NOT_TRUSTED:
                 return TLSDetails(domain_name=domain, error_message="invalid: root not trusted.")
             else:
                 return TLSDetails(domain_name=domain, error_message="failed verification: " + e.verify_message + ".")
         except ssl.SSLError as e:
             return TLSDetails(domain_name=domain, error_message="could not establish a secure connection: " + e.reason + ".")
         except Exception as e:
             return TLSDetails(domain_name=domain, error_message="could not connect: " + str(e) + ".")
@@ \ No newline at end of file @@

backend/mail.py

➞

Show inline comments

@@ @@ -5,15 +5,10 @@ import imaplib @@
 from rich.console import Console
 from cryptography import x509
 import tls_utils
-from tls_utils import TLSDetails
+from generic_handler import GenericHandler
 from abc import ABC, abstractmethod
 class MailHandler(ABC):
     def __init__(self, host: str, port: int, context: ssl.SSLContext):
         self.host = host
         self.port = port
         self.context = context
 class MailHandler(GenericHandler):
     def connect(self, verification: bool) -> int:
         connection = self.protocol_init(self.host, self.port)
         if verification:
@@ @@ -22,7 +17,7 @@ class MailHandler(ABC): @@
             connection.starttls()
         cert = connection.sock.getpeercert()
         self.protocol_close(connection)
-        return tls_utils.get_validity_days(cert)[1]
+        return tls_utils.check_cert_validity(cert)[1]
     @abstractmethod
     def protocol_init(self, host, port):
@@ @@ -34,15 +29,6 @@ class MailHandler(ABC): @@
     def protocol_starttls_args(self):
         raise NotImplementedError()
     @staticmethod
     def create_handler(protocol: str):
         if protocol == "smtp":
             return SMTPHandler
         elif protocol == "imap":
             return IMAPHandler
         else:
             raise ValueError("Invalid protocol")
 class IMAPHandler(MailHandler):
     def protocol_init(self, host, port):
         return imaplib.IMAP4(host, port)
@@ @@ -57,21 +43,4 @@ class SMTPHandler(MailHandler): @@
     def protocol_close(self, connection):
         connection.quit()
     def protocol_starttls_args(self):
         return {"context": self.context}
 class MailVerificator:
     def __init__(self, context: ssl.SSLContext):
         self.context = context
     def connect(self, domain: str, port: int, protocol: str) -> TLSDetails:
         mail = MailHandler.create_handler(protocol)(domain, port, self.context)
         try:
             expiry = mail.connect(True)
             return TLSDetails(domain_name=domain, expires_in_days=expiry)
         except ssl.SSLCertVerificationError as e:
             if (e.verify_code == tls_utils.EXPIRED_VERIFY_CODE):
                 expiry = mail.connect(False)
                 return TLSDetails(domain_name=domain, expires_in_days=expiry)
             else:
                 error = "failed verification:", e.verify_message + "."
                 return TLSDetails(domain_name=domain, error_message=error)
@@ \ No newline at end of file @@
         return {"context": self.context}
@@ \ No newline at end of file @@

backend/tls_utils.py

➞

Show inline comments

@@ @@ -30,17 +30,31 @@ class TLSDetails: @@
         else:
             console.log("[green bold underline]" + self.domain_name, "expires in", self.expires_in_days, "days", style="green")
-def get_expiry_timestamps(expiry_timestamp: int, now_timestamp: int = datetime.datetime.now(datetime.UTC).timestamp()) -> tuple[bool, int]:
+def compare_expiry_timestamps(expiry_timestamp: int, now_timestamp: int = datetime.datetime.now(datetime.UTC).timestamp()) -> tuple[bool, int]:
     seconds_left = expiry_timestamp - now_timestamp
     valid = seconds_left >= 0
     # We use floor(), which, when negative, will round towards -1
     if not valid:
         seconds_left = -seconds_left
     days_left = math.floor(seconds_left / 86400)
     return (seconds_left >= 0, days_left)
     # We need to restore the inversion
     if not valid:
         days_left = -days_left
     return (valid, days_left)
 def get_validity_days(cert) -> tuple[bool, int]:
 # Returns if the cert is valid, and the number of days left until expiry (negative if expired)
 def check_cert_validity(cert) -> tuple[bool, int]:
     # Get expiry date
     notAfter = cert['notAfter']
     notAfter_date = datetime.datetime.strptime(notAfter, '%b %d %H:%M:%S %Y %Z')
     # datetime to UNIX time
     notAfter_timestamp = notAfter_date.timestamp()
     expiry = get_expiry_timestamps(notAfter_timestamp)
     return (expiry[0], abs(expiry[1]))
@@ \ No newline at end of file @@
     expiry = compare_expiry_timestamps(notAfter_timestamp)
     return (expiry[0], expiry[1])
 # Test expiry checking (timestamps)
 if __name__ == "__main__":
     console = Console()
     console.log("Time from rn (some time ago):", compare_expiry_timestamps(1715277129))
     console.log("Time from rn (in some time):", compare_expiry_timestamps(1715279129))
@@ \ No newline at end of file @@

backend/web.py

➞

Show inline comments

 #!/usr/bin/env python3
 import ssl
 from rich.console import Console
 from cryptography import x509
 import socket
 import tls_utils
 from tls_utils import TLSDetails
 class SSLHandler:
     def __init__(self, host: str, port: int, context: ssl.SSLContext):
         self.host = host
         self.port = port
         self.context = context
 from generic_handler import GenericHandler
 class SSLHandler(GenericHandler):
     def connect(self, verification: bool) -> int:
         if verification:
             with self.context.wrap_socket(socket.socket(), server_hostname=self.host) as s:
                 s.connect((self.host, self.port))
                 cert = s.getpeercert()
-                return tls_utils.get_validity_days(cert)[1]
+                return tls_utils.check_cert_validity(cert)[1]
         else:
             pem_cert = ssl.get_server_certificate((self.host, self.port), timeout=5)
             cert = x509.load_pem_x509_certificate(pem_cert.encode())
             not_after = cert.not_valid_after_utc.timestamp()
             return tls_utils.get_expiry_timestamps(not_after)[1]
 class SSLVerificator:
     def __init__(self, context: ssl.SSLContext):
         self.context = context
     def connect(self, domain: str, port: int) -> TLSDetails:
         handler = SSLHandler(domain, port, self.context)
         try:
             expiry = handler.connect(True)
             return TLSDetails(domain_name=domain, expires_in_days=expiry)
         except ssl.SSLCertVerificationError as e:
             if e.verify_code == tls_utils.EXPIRED:
                 expiry = handler.connect(False)
                 return TLSDetails(domain_name=domain, expires_in_days=expiry)
             elif e.verify_code == tls_utils.REVOKED:
                 return TLSDetails(domain_name=domain, error_message="was revoked.")
             elif e.verify_code == tls_utils.SELF_SIGNED:
                 return TLSDetails(domain_name=domain, error_message="is self-signed.")
             elif e.verify_code == tls_utils.ROOT_NOT_TRUSTED:
                 return TLSDetails(domain_name=domain, error_message="invalid: root not trusted.")
             else:
                 return TLSDetails(domain_name=domain, error_message="failed verification: " + e.verify_message + ".")
         except ssl.SSLError as e:
             return TLSDetails(domain_name=domain, error_message="could not establish a secure connection: " + e.reason + ".")
         except Exception as e:
             return TLSDetails(domain_name=domain, error_message="could not connect: " + str(e) + ".")
@@ \ No newline at end of file @@
             return tls_utils.compare_expiry_timestamps(not_after)[1]
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)