Merge commit 'a1c1853ca54479c5f0ac649db2ab55cf219336cd' as 'utils_v2'
This commit is contained in:
@@ -0,0 +1,481 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Friday, 30th May, 2025
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a way to analyze web certs.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# System-level activities:
|
||||
import io
|
||||
|
||||
# To check certs through cryptographic algorithms:
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.x509.oid import NameOID
|
||||
|
||||
# To fetch remote certs:
|
||||
import ssl
|
||||
import socket
|
||||
from urllib.parse import urlparse
|
||||
|
||||
# To work with date and time:
|
||||
import datetime
|
||||
import humanize
|
||||
|
||||
# For debugging:
|
||||
from icecream import IceCreamDebugger
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** CLASSES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class CertAnalyzer:
|
||||
|
||||
# Certificate variables:
|
||||
cert_path: str | None = None
|
||||
cert_data = None
|
||||
serial_no: int | None = None
|
||||
signature: bytes | None = None
|
||||
signature_algorithm: str | None = None
|
||||
not_valid_before: datetime.datetime | None = None
|
||||
not_valid_after: datetime.datetime | None = None
|
||||
subject: dict | None = None
|
||||
issuer: dict | None = None
|
||||
extensions: dict | None = None
|
||||
public_key_type: str | None = None
|
||||
public_key_size: int | None = None
|
||||
public_key: bytes | None = None
|
||||
exception: Exception | None = None
|
||||
exception_str: str | None = None
|
||||
|
||||
# ┏┓
|
||||
# ┃ ┏┓┏┓╋┏┓┓┏┏╋┏┓┏┓
|
||||
# ┗┛┗┛┛┗┗┛ ┗┻┗┗┗┛┛
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
debug: bool = False,
|
||||
debug_prefix: str = "Certs Exp. | "
|
||||
):
|
||||
|
||||
"""
|
||||
This simple class just checks if a provided certificate is valid or not and what is the date range in which it
|
||||
is valid.
|
||||
"""
|
||||
|
||||
# Prepare the debugging utility:
|
||||
self._debug_prefix = debug_prefix
|
||||
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
|
||||
if not debug: self._printer.disable()
|
||||
|
||||
# ┓┏ ┓
|
||||
# ┣┫┏┓┃┏┓┏┓┏┓┏
|
||||
# ┛┗┗ ┗┣┛┗ ┛ ┛
|
||||
# ┛
|
||||
|
||||
def enable_debug(self):
|
||||
self._printer.enable()
|
||||
|
||||
def disable_debug(self):
|
||||
self._printer.disable()
|
||||
|
||||
def _clear(self):
|
||||
self.cert_path = None
|
||||
self.cert_data = None
|
||||
self.serial_no = None
|
||||
self.signature = None
|
||||
self.signature_algorithm = None
|
||||
self.not_valid_before = None
|
||||
self.not_valid_after = None
|
||||
self.subject = None
|
||||
self.issuer = None
|
||||
self.extensions = None
|
||||
self.public_key_type = None
|
||||
self.public_key_size = None
|
||||
self.public_key = None
|
||||
self.exception = None
|
||||
self.exception_str = None
|
||||
|
||||
# ┏┓ •
|
||||
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
|
||||
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
|
||||
# ┛
|
||||
|
||||
@property
|
||||
def is_valid(self) -> bool | None:
|
||||
|
||||
"""
|
||||
Quickly tests if a loaded certificate is valid or not.
|
||||
:return: True if the certificate is valid, False if it isn't. Returns None if no certificate was loaded.
|
||||
"""
|
||||
|
||||
# Check if the certificate was loaded correctly:
|
||||
if self.not_valid_after is None or self.not_valid_before is None:
|
||||
self._printer("Please load a certificate first.")
|
||||
return None
|
||||
|
||||
# Check if the cert is within the given time limits:
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
if self.not_valid_before <= now <= self.not_valid_after: return True
|
||||
else: return False
|
||||
|
||||
@property
|
||||
def ttl(self) -> float | None:
|
||||
|
||||
"""
|
||||
Gives the time for which the certificate is valid in seconds.
|
||||
:return: Time for which the certificate is valid in seconds. None if the certificate was not loaded.
|
||||
"""
|
||||
|
||||
# Check if the certificate was loaded correctly:
|
||||
if self.not_valid_after is None or self.not_valid_before is None:
|
||||
self._printer("Please load a certificate first.")
|
||||
return None
|
||||
|
||||
# Calculate the no. of seconds left till expiry:
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
ttl_seconds = (self.not_valid_after - now).total_seconds()
|
||||
return ttl_seconds
|
||||
|
||||
@property
|
||||
def ttl_human_readable(self) -> str | None:
|
||||
|
||||
"""
|
||||
Gives human-readable validity period for certificates.
|
||||
e.g.: 1. valid for a month
|
||||
2. expired 3 days ago
|
||||
3. valid for 2 months
|
||||
:return: Human-readable validity period for certificates. None if the certificate was not loaded.
|
||||
"""
|
||||
|
||||
# Check if the certificate was loaded correctly:
|
||||
if self.not_valid_after is None or self.not_valid_before is None:
|
||||
self._printer("Please load a certificate first.")
|
||||
return None
|
||||
|
||||
# Calculate the no. of seconds left till expiry:
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
ttl_seconds = (self.not_valid_after - now).total_seconds()
|
||||
ttl_readable = humanize.naturaldelta(datetime.timedelta(seconds = ttl_seconds))
|
||||
if ttl_seconds <= 0: ttl_readable = f"expired {ttl_readable} ago"
|
||||
else: ttl_readable = f"valid for {ttl_readable}"
|
||||
return ttl_readable
|
||||
|
||||
@property
|
||||
def json(self) -> dict | None:
|
||||
|
||||
"""
|
||||
Gives a JSON dump (as a dict) of the contents of the certificate.
|
||||
:return: JSON dump of the contents of the certificate; None if no certificate was loaded.
|
||||
"""
|
||||
|
||||
# Gives nothing if the certificate failed to load:
|
||||
if self.cert_data is None:
|
||||
return None
|
||||
|
||||
# Otherwise, return a formatted dictionary if teh certificate:
|
||||
return {
|
||||
"path": self.cert_path,
|
||||
"data": self.cert_data,
|
||||
"serialNo": self.serial_no,
|
||||
"signature": self.signature,
|
||||
"signatureAlgorithm": self.signature_algorithm,
|
||||
"notValidBefore": self.not_valid_before,
|
||||
"notValidAfter": self.not_valid_after,
|
||||
"isValid": self.is_valid,
|
||||
"ttl": self.ttl,
|
||||
"ttlHumanReadable": self.ttl_human_readable,
|
||||
"subject": self.subject,
|
||||
"issuer": self.issuer,
|
||||
"extensions": self.extensions,
|
||||
"publicKeyType": self.public_key_type,
|
||||
"publicKeySize": self.public_key_size,
|
||||
"publicKey": self.public_key,
|
||||
}
|
||||
|
||||
# ┏┓ ┓ •
|
||||
# ┣┫┏┓┏┓┃┓┏┏┓┏
|
||||
# ┛┗┛┗┗┻┗┗┫┛┗┛
|
||||
# ┛
|
||||
|
||||
def _analyze_cert(self) -> bool:
|
||||
|
||||
"""
|
||||
This method analyzes the certificate that was provided either as a local file or a remote file. Call this from
|
||||
other methods that load certs (once the cert has been loaded by them.
|
||||
:return: True if the certificate has been analyzed, False otherwise.
|
||||
"""
|
||||
|
||||
# Check if a cert was loaded:
|
||||
if self.cert_data is None:
|
||||
return False
|
||||
|
||||
# Get the serial number and signature details of the cert:
|
||||
self.serial_no = self.cert_data.serial_number
|
||||
self.signature = self.cert_data.signature
|
||||
# self.signature_algorithm = self.cert_data.signature_hash_algorithm.name
|
||||
self.signature_algorithm = self.cert_data.signature_algorithm_oid._name
|
||||
|
||||
# Get the subject:
|
||||
_sub = self.cert_data.subject
|
||||
self.subject = {
|
||||
"CN": _sub.get_attributes_for_oid(NameOID.COMMON_NAME),
|
||||
"O": _sub.get_attributes_for_oid(NameOID.ORGANIZATION_NAME),
|
||||
"OU": _sub.get_attributes_for_oid(NameOID.ORGANIZATIONAL_UNIT_NAME),
|
||||
"C": _sub.get_attributes_for_oid(NameOID.COUNTRY_NAME),
|
||||
"ST": _sub.get_attributes_for_oid(NameOID.STATE_OR_PROVINCE_NAME),
|
||||
"L": _sub.get_attributes_for_oid(NameOID.LOCALITY_NAME),
|
||||
"emailAddress": _sub.get_attributes_for_oid(NameOID.EMAIL_ADDRESS),
|
||||
}
|
||||
for k in self.subject.keys():
|
||||
v = self.subject[k]
|
||||
self.subject[k] = v[0].value if v else None
|
||||
|
||||
# Get the issuer:
|
||||
_iss = self.cert_data.issuer
|
||||
self.issuer = {
|
||||
"CN": _iss.get_attributes_for_oid(NameOID.COMMON_NAME),
|
||||
"O": _iss.get_attributes_for_oid(NameOID.ORGANIZATION_NAME),
|
||||
"OU": _iss.get_attributes_for_oid(NameOID.ORGANIZATIONAL_UNIT_NAME),
|
||||
"C": _iss.get_attributes_for_oid(NameOID.COUNTRY_NAME),
|
||||
"ST": _iss.get_attributes_for_oid(NameOID.STATE_OR_PROVINCE_NAME),
|
||||
"L": _iss.get_attributes_for_oid(NameOID.LOCALITY_NAME),
|
||||
"emailAddress": _iss.get_attributes_for_oid(NameOID.EMAIL_ADDRESS),
|
||||
}
|
||||
for k in self.issuer.keys():
|
||||
v = self.issuer[k]
|
||||
self.issuer[k] = v[0].value if v else None
|
||||
|
||||
# # Get the extensions:
|
||||
# _ext = self.cert_data.extensions
|
||||
# for ext in self.cert_data.extensions:
|
||||
# print(f"Extension OID: {ext.oid._name if ext.oid._name else ext.oid.dotted_string}")
|
||||
# print(f"Critical: {ext.critical}")
|
||||
# print(f"Value: {ext.value}")
|
||||
# print("-" * 10)
|
||||
# self.extensions = {
|
||||
#
|
||||
# }
|
||||
|
||||
# Get the not valid before and not valid after dates:
|
||||
self.not_valid_before = self.cert_data.not_valid_before_utc
|
||||
self.not_valid_after = self.cert_data.not_valid_after_utc
|
||||
|
||||
# Extract the key, and its type and size:
|
||||
_public_key = self.cert_data.public_key()
|
||||
self.public_key_type = type(_public_key).__name__
|
||||
self.public_key_size = _public_key.key_size
|
||||
self.public_key = _public_key.public_bytes(
|
||||
encoding = serialization.Encoding.PEM,
|
||||
format = serialization.PublicFormat.SubjectPublicKeyInfo
|
||||
)
|
||||
|
||||
# Get the not valid before and not valid after dates:
|
||||
self.not_valid_before = self.cert_data.not_valid_before_utc
|
||||
self.not_valid_after = self.cert_data.not_valid_after_utc
|
||||
|
||||
# Done here:
|
||||
return True
|
||||
|
||||
# ┓ ┓ ┏┓
|
||||
# ┃ ┏┓┏┏┓┃ ┃ ┏┓┏┓╋┏
|
||||
# ┗┛┗┛┗┗┻┗ ┗┛┗ ┛ ┗┛
|
||||
|
||||
def load_local(self, path: str | io.BytesIO) -> bool:
|
||||
|
||||
"""
|
||||
Loads a certificate stored on a local file path and checks for its validity.
|
||||
:param path: The local file path where the certificate is stored or a file in RAM.
|
||||
:return: True if the certificate was loaded correctly, False if the certificate wasn't loaded.
|
||||
"""
|
||||
|
||||
# Clear any previous assessment:
|
||||
self._clear()
|
||||
|
||||
# Note down the certificate that is being tested:
|
||||
self.cert_path = path
|
||||
|
||||
try:
|
||||
|
||||
# Read the contents of the certificate file:
|
||||
if isinstance(path, str):
|
||||
with open(path, "rb") as f: cert_data = f.read()
|
||||
elif isinstance(path, io.BytesIO):
|
||||
path.seek(0)
|
||||
cert_data = path.read()
|
||||
else: cert_data = None
|
||||
|
||||
# Load the certificate:
|
||||
self.cert_data = x509.load_pem_x509_certificate(cert_data, default_backend())
|
||||
|
||||
# Analyze the certificate:
|
||||
self._analyze_cert()
|
||||
|
||||
# All done successfully:
|
||||
return True
|
||||
|
||||
except Exception as exception:
|
||||
self._printer(path, exception)
|
||||
self.exception = exception
|
||||
self.exception_str = str(exception)
|
||||
self.cert_data = None
|
||||
return False
|
||||
|
||||
# ┳┓ ┏┓
|
||||
# ┣┫┏┓┏┳┓┏┓╋┏┓ ┃ ┏┓┏┓╋┏
|
||||
# ┛┗┗ ┛┗┗┗┛┗┗ ┗┛┗ ┛ ┗┛
|
||||
|
||||
def load_remote(self, hostname: str, port: int = 443) -> bool:
|
||||
|
||||
"""
|
||||
Loads the certificate of a remote host and checks for its validity.
|
||||
:param hostname: The remote host URL.
|
||||
:param port: The remote host's port on which the service is running.
|
||||
:return: True if the certificate was loaded correctly, False if the certificate wasn't loaded.
|
||||
"""
|
||||
|
||||
# Clear any previous assessment:
|
||||
self._clear()
|
||||
|
||||
# Parse the URL's hostname:
|
||||
hostname = urlparse(hostname).hostname or hostname
|
||||
|
||||
# Note down the certificate that is being tested:
|
||||
self.cert_path = f"{hostname}:{port}"
|
||||
|
||||
try:
|
||||
|
||||
# Fetch the remote certificate:
|
||||
context = ssl.create_default_context()
|
||||
with socket.create_connection((hostname, port)) as sock:
|
||||
with context.wrap_socket(sock, server_hostname = hostname) as ssock:
|
||||
|
||||
# Attempt to retrieve the certificate in DER format
|
||||
try:
|
||||
der_cert = ssock.getpeercert(binary_form = True)
|
||||
self.cert_data = x509.load_der_x509_certificate(der_cert, default_backend())
|
||||
|
||||
# If DER parsing fails, retrieve the certificate in PEM format:
|
||||
except ValueError:
|
||||
pem_cert = ssl.DER_cert_to_PEM_cert(ssock.getpeercert(binary_form = True))
|
||||
self.cert_data = x509.load_pem_x509_certificate(pem_cert.encode(), default_backend())
|
||||
|
||||
# Analyze the certificate:
|
||||
self._analyze_cert()
|
||||
|
||||
# All done successfully:
|
||||
return True
|
||||
|
||||
except Exception as exception:
|
||||
self._printer(hostname, port, exception)
|
||||
self.exception = exception
|
||||
self.exception_str = str(exception)
|
||||
self.cert_data = None
|
||||
return False
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
def show_details(my_cert: CertAnalyzer):
|
||||
print("Cert. Path :", my_cert.cert_path)
|
||||
print("Serial No. :", my_cert.serial_no)
|
||||
print("Signature :", my_cert.signature)
|
||||
print("Sig. Algo. :", my_cert.signature_algorithm)
|
||||
print("Not Before :", my_cert.not_valid_before)
|
||||
print("Not After :", my_cert.not_valid_after)
|
||||
print("Is Valid :", my_cert.is_valid)
|
||||
print("TTL (seconds) :", my_cert.ttl)
|
||||
print("TTL (human) :", my_cert.ttl_human_readable)
|
||||
print("Subject :", my_cert.subject)
|
||||
print("Issuer :", my_cert.issuer)
|
||||
print("Pub. Key Type :", my_cert.public_key_type)
|
||||
print("Pub. Key Size :", my_cert.public_key_size)
|
||||
print("Pub. Key :", my_cert.public_key)
|
||||
print("Exception :", my_cert.exception_str)
|
||||
print("JSON Dump :", my_cert.json)
|
||||
|
||||
# Create an instance:
|
||||
my_cert = CertAnalyzer(debug = False)
|
||||
|
||||
# Test a remote path:
|
||||
print("\n---\n")
|
||||
my_cert.load_remote("https://thecaoffice.com/", port = 443)
|
||||
show_details(my_cert)
|
||||
|
||||
# Test a local path:
|
||||
print("\n---\n")
|
||||
my_cert.load_local(input("Cert File Path (Local): "))
|
||||
print("----------------------")
|
||||
show_details(my_cert)
|
||||
@@ -0,0 +1,340 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Friday, 30th May, 2025
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a base class for common behaviour of Google's APIs.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
1. Quickstart: https://developers.google.com/gmail/api/quickstart/python
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# System-level activities:
|
||||
import io
|
||||
|
||||
# To check certs through cryptographic algorithms:
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
|
||||
# To fetch remote certs:
|
||||
import ssl
|
||||
import socket
|
||||
from urllib.parse import urlparse
|
||||
|
||||
# To work with date and time:
|
||||
import datetime
|
||||
import humanize
|
||||
|
||||
# For debugging:
|
||||
from icecream import IceCreamDebugger
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** CLASSES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class CertExpiryCheck:
|
||||
|
||||
# Certificate variables:
|
||||
certificate: str | None = None
|
||||
not_valid_before: datetime.datetime | None = None
|
||||
not_valid_after: datetime.datetime | None = None
|
||||
exception: Exception | None = None
|
||||
exception_str: str | None = None
|
||||
|
||||
# ┏┓
|
||||
# ┃ ┏┓┏┓╋┏┓┓┏┏╋┏┓┏┓
|
||||
# ┗┛┗┛┛┗┗┛ ┗┻┗┗┗┛┛
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
debug: bool = False,
|
||||
debug_prefix: str = "Certs Exp. | "
|
||||
):
|
||||
|
||||
"""
|
||||
This simple class just checks if a provided certificate is valid or not and what is the date range in which it
|
||||
is valid.
|
||||
"""
|
||||
|
||||
# Prepare the debugging utility:
|
||||
self._debug_prefix = debug_prefix
|
||||
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
|
||||
if not debug: self._printer.disable()
|
||||
|
||||
# ┓┏ ┓
|
||||
# ┣┫┏┓┃┏┓┏┓┏┓┏
|
||||
# ┛┗┗ ┗┣┛┗ ┛ ┛
|
||||
# ┛
|
||||
|
||||
def enable_debug(self):
|
||||
self._printer.enable()
|
||||
|
||||
def disable_debug(self):
|
||||
self._printer.disable()
|
||||
|
||||
def _clear(self):
|
||||
self.not_valid_before = None
|
||||
self.not_valid_after = None
|
||||
self.exception = None
|
||||
self.exception_str = None
|
||||
|
||||
# ┏┓ •
|
||||
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
|
||||
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
|
||||
# ┛
|
||||
|
||||
@property
|
||||
def is_valid(self) -> bool | None:
|
||||
|
||||
"""
|
||||
Quickly tests if a loaded certificate is valid or not.
|
||||
:return: True if the certificate is valid, False if it isn't. Returns None if no certificate was loaded.
|
||||
"""
|
||||
|
||||
# Check if the certificate was loaded correctly:
|
||||
if self.not_valid_after is None or self.not_valid_before is None:
|
||||
self._printer("Please load a certificate first.")
|
||||
return None
|
||||
|
||||
# Check if the cert is within the given time limits:
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
if self.not_valid_before <= now <= self.not_valid_after: return True
|
||||
else: return False
|
||||
|
||||
@property
|
||||
def ttl(self) -> float | None:
|
||||
|
||||
"""
|
||||
Gives the time for which the certificate is valid in seconds.
|
||||
:return: Time for which the certificate is valid in seconds. None if the certificate was not loaded.
|
||||
"""
|
||||
|
||||
# Check if the certificate was loaded correctly:
|
||||
if self.not_valid_after is None or self.not_valid_before is None:
|
||||
self._printer("Please load a certificate first.")
|
||||
return None
|
||||
|
||||
# Calculate the no. of seconds left till expiry:
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
ttl_seconds = (self.not_valid_after - now).total_seconds()
|
||||
return ttl_seconds
|
||||
|
||||
@property
|
||||
def ttl_human_readable(self) -> str | None:
|
||||
|
||||
"""
|
||||
Gives human-readable validity period for certificates.
|
||||
e.g.: 1. valid for a month
|
||||
2. expired 3 days ago
|
||||
3. valid for 2 months
|
||||
:return: Human-readable validity period for certificates. None if the certificate was not loaded.
|
||||
"""
|
||||
|
||||
# Check if the certificate was loaded correctly:
|
||||
if self.not_valid_after is None or self.not_valid_before is None:
|
||||
self._printer("Please load a certificate first.")
|
||||
return None
|
||||
|
||||
# Calculate the no. of seconds left till expiry:
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
ttl_seconds = (self.not_valid_after - now).total_seconds()
|
||||
ttl_readable = humanize.naturaldelta(datetime.timedelta(seconds = ttl_seconds))
|
||||
if ttl_seconds <= 0: ttl_readable = f"expired {ttl_readable} ago"
|
||||
else: ttl_readable = f"valid for {ttl_readable}"
|
||||
return ttl_readable
|
||||
|
||||
# ┓ ┓ ┏┓
|
||||
# ┃ ┏┓┏┏┓┃ ┃ ┏┓┏┓╋┏
|
||||
# ┗┛┗┛┗┗┻┗ ┗┛┗ ┛ ┗┛
|
||||
|
||||
def load_local(self, path: str | io.BytesIO) -> bool:
|
||||
|
||||
"""
|
||||
Loads a certificate stored on a local file path and checks for its validity.
|
||||
:param path: local file path where the certificate is stored.
|
||||
:return: True if the certificate was loaded correctly, False if the certificate wasn't loaded.
|
||||
"""
|
||||
|
||||
# Note down the certificate that is being tested:
|
||||
self.certificate = path
|
||||
|
||||
# Clear any previous assessment:
|
||||
self._clear()
|
||||
|
||||
try:
|
||||
|
||||
# Read the contents of the certificate file:
|
||||
if isinstance(path, str):
|
||||
with open(path, 'rb') as f:
|
||||
cert_data = f.read()
|
||||
elif isinstance(path, io.BytesIO):
|
||||
path.seek(0)
|
||||
cert_data = path.read()
|
||||
else: cert_data = None
|
||||
|
||||
# Load the certificate:
|
||||
cert = x509.load_pem_x509_certificate(cert_data, default_backend())
|
||||
|
||||
# Get the not valid before and not valid after dates:
|
||||
self.not_valid_before = cert.not_valid_before_utc
|
||||
self.not_valid_after = cert.not_valid_after_utc
|
||||
|
||||
# All done successfully:
|
||||
return True
|
||||
|
||||
except Exception as exception:
|
||||
self._printer(path, exception)
|
||||
self.exception = exception
|
||||
self.exception_str = str(exception)
|
||||
return False
|
||||
|
||||
# ┳┓ ┏┓
|
||||
# ┣┫┏┓┏┳┓┏┓╋┏┓ ┃ ┏┓┏┓╋┏
|
||||
# ┛┗┗ ┛┗┗┗┛┗┗ ┗┛┗ ┛ ┗┛
|
||||
|
||||
def load_remote(self, hostname: str, port: int = 443) -> bool:
|
||||
|
||||
"""
|
||||
Loads the certificate of a remote host and checks for its validity.
|
||||
:param hostname: The remote host URL.
|
||||
:param port: The remote host's port on which the service is running.
|
||||
:return: True if the certificate was loaded correctly, False if the certificate wasn't loaded.
|
||||
"""
|
||||
|
||||
# Parse the URL's hostname:
|
||||
hostname = urlparse(hostname).hostname or hostname
|
||||
|
||||
# Note down the certificate that is being tested:
|
||||
self.certificate = f"{hostname}:{port}"
|
||||
|
||||
# Clear any previous assessment:
|
||||
self._clear()
|
||||
|
||||
try:
|
||||
|
||||
# Fetch the remote certificate:
|
||||
context = ssl.create_default_context()
|
||||
with socket.create_connection((hostname, port)) as sock:
|
||||
with context.wrap_socket(sock, server_hostname = hostname) as ssock:
|
||||
|
||||
# Attempt to retrieve the certificate in DER format
|
||||
try:
|
||||
der_cert = ssock.getpeercert(binary_form = True)
|
||||
cert = x509.load_der_x509_certificate(der_cert, default_backend())
|
||||
|
||||
# If DER parsing fails, retrieve the certificate in PEM format:
|
||||
except ValueError:
|
||||
pem_cert = ssl.DER_cert_to_PEM_cert(ssock.getpeercert(binary_form = True))
|
||||
cert = x509.load_pem_x509_certificate(pem_cert.encode(), default_backend())
|
||||
|
||||
# Get the not valid before and not valid after dates:
|
||||
self.not_valid_before = cert.not_valid_before_utc
|
||||
self.not_valid_after = cert.not_valid_after_utc
|
||||
|
||||
# All done successfully:
|
||||
return True
|
||||
|
||||
except Exception as exception:
|
||||
self._printer(hostname, port, exception)
|
||||
self.exception = exception
|
||||
self.exception_str = str(exception)
|
||||
return False
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
def show_details(my_cert: CertExpiryCheck):
|
||||
print("\n---\n")
|
||||
print("Certificate :", my_cert.certificate)
|
||||
print("Not Before :", my_cert.not_valid_before)
|
||||
print("Not After :", my_cert.not_valid_after)
|
||||
print("Is Valid :", my_cert.is_valid)
|
||||
print("TTL (seconds) :", my_cert.ttl)
|
||||
print("TTL (human) :", my_cert.ttl_human_readable)
|
||||
print("Exception :", my_cert.exception_str)
|
||||
|
||||
# Create an instance:
|
||||
my_cert = CertExpiryCheck(debug = False)
|
||||
|
||||
# Test a local path:
|
||||
my_cert.load_local(r"C:\Users\Khushal P Soonderji\Desktop\Dktp Files\certs\mongo\mongo_data_cert.pem")
|
||||
show_details(my_cert)
|
||||
|
||||
# Test a local path:
|
||||
my_cert.load_local(r"C:\Users\Khushal P Soonderji\Desktop\Dktp Files\certs\kafka\privkey.pem")
|
||||
show_details(my_cert)
|
||||
|
||||
# Test a local path:
|
||||
my_cert.load_local(r"C:\Users\Khushal P Soonderji\Desktop\Dktp Files\certs\kafka\fullchain.pem")
|
||||
show_details(my_cert)
|
||||
|
||||
# Test a remote path:
|
||||
my_cert.load_remote("https://thecaoffice.com/", port = 443)
|
||||
show_details(my_cert)
|
||||
Reference in New Issue
Block a user