tanner/ssl-monitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Log warnings for duplicate certificates per protocol

Browse Source 
Co-authored-by: aider (gemini/gemini-2.5-pro) <aider@aider.chat>
This commit is contained in:
Tanner Collin
2026-01-14 10:45:00 -07:00
parent 476342d654
commit 1838cf4c7b
1 changed files with 11 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
main.py
View File

@@ -23,7 +23,7 @@ except FileNotFoundError:
def alert_tanner(msg): def alert_tanner(msg):
pass pass
async def check_host_cert(host, port): async def check_host_cert(host, port, seen_serials):
"check a single host's cert" "check a single host's cert"
try: try:
# default context does hostname checking and certificate validation # default context does hostname checking and certificate validation
@@ -43,6 +43,13 @@ async def check_host_cert(host, port):
alert_tanner(msg) alert_tanner(msg)
return return
serial_number = cert.get('serialNumber')
if serial_number:
if serial_number in seen_serials:
logging.warning(f"Duplicate certificate with serial number {serial_number} found for host {host}:{port}")
else:
seen_serials.add(serial_number)
expiry_date_str = cert['notAfter'] expiry_date_str = cert['notAfter']
expiry_date = datetime.strptime(expiry_date_str, '%b %d %H:%M:%S %Y %Z') expiry_date = datetime.strptime(expiry_date_str, '%b %d %H:%M:%S %Y %Z')
@@ -74,13 +81,14 @@ async def check_host_cert(host, port):
async def main(): async def main():
seen_serials = {proto: set() for proto in HOSTS}
tasks = [] tasks = []
for host in HOSTS['http']: for host in HOSTS['http']:
tasks.append(check_host_cert(host, 443)) tasks.append(check_host_cert(host, 443, seen_serials['http']))
for host in HOSTS['mqtt']: for host in HOSTS['mqtt']:
# standard port for MQTTS is 8883 # standard port for MQTTS is 8883
tasks.append(check_host_cert(host, 8883)) tasks.append(check_host_cert(host, 8883, seen_serials['mqtt']))
await asyncio.gather(*tasks) await asyncio.gather(*tasks)