""" SSL Certificate Monitor - Daily Automated SSL Checking This module provides automated daily SSL certificate monitoring for all active user websites. It checks SSL certificates, updates the database, and sends alert emails when certificates are expired or expiring soon. """ import asyncio from datetime import datetime, timezone, timedelta from typing import Dict, Any, List, Optional from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from bson import ObjectId import os from reports.database import db from reports.ssl_checker import check_ssl_certificate_async from reports.email_sender import send_ssl_alert_email # Configuration SSL_CHECK_TIME_HOUR = 0 # Run daily at 12:00 AM (midnight) - start of new day SSL_CHECK_TIME_MINUTE = 0 # At minute 0 SSL_ALERT_THRESHOLD_DAYS = 7 # Send alert if expiring within 7 days SSL_URGENT_THRESHOLD_DAYS = 3 # Send urgent alert if expiring within 3 days class SSLMonitor: """ Automated SSL certificate monitoring service. Features: - Daily SSL checks for all active user websites - Automatic database updates with SSL status - Email alerts for expired/expiring certificates - Prevents duplicate alerts within 24 hours """ def __init__(self): self.scheduler = AsyncIOScheduler() self.is_running = False self.last_check_stats = { "total_websites": 0, "successful_checks": 0, "failed_checks": 0, "alerts_sent": 0, "last_run": None } def start(self): """ Start the SSL monitoring scheduler. Runs daily at midnight (00:00) - start of new day. """ if self.is_running: print("SSL Monitor is already running") return # Schedule daily SSL check at midnight (start of new day) self.scheduler.add_job( func=self.run_daily_ssl_check, trigger=CronTrigger(hour=SSL_CHECK_TIME_HOUR, minute=SSL_CHECK_TIME_MINUTE), id="daily_ssl_check", name="Daily SSL Certificate Check", replace_existing=True ) self.scheduler.start() self.is_running = True print(f"SSL Monitor started - Daily checks at {SSL_CHECK_TIME_HOUR:02d}:{SSL_CHECK_TIME_MINUTE:02d} (midnight - start of new day)") def stop(self): """Stop the SSL monitoring scheduler.""" if self.scheduler.running: self.scheduler.shutdown() self.is_running = False print("SSL Monitor stopped") async def run_daily_ssl_check(self): """ Main method that runs daily SSL checks for all active websites. Process: 1. Fetch all active users with verified emails 2. Get all unique websites from reports collection 3. Check SSL certificate for each website 4. Update database with SSL status 5. Send email alerts if certificate is expired or expiring soon """ print(f"\nStarting daily SSL check at {datetime.utcnow().isoformat()}") # Reset stats self.last_check_stats = { "total_websites": 0, "successful_checks": 0, "failed_checks": 0, "alerts_sent": 0, "last_run": datetime.utcnow() } try: # Step 1: Get all active, verified users active_users = await db["users"].find( { "is_verified": True, "status": "active" }, {"_id": 1, "email": 1, "username": 1, "websites": 1} ).to_list(length=None) if not active_users: print("No active users found for SSL monitoring") return print(f"Found {len(active_users)} active users") # Step 2: Collect all unique websites from user dashboards websites_to_check = set() user_website_map = {} # Map website -> list of users for user in active_users: user_websites = user.get("websites", []) if user_websites: for website in user_websites: # Normalize URL (remove trailing slash) website = website.rstrip("/") websites_to_check.add(website) # Track which users own this website if website not in user_website_map: user_website_map[website] = [] user_website_map[website].append(user) # Also get websites from recent reports (last 90 days) ninety_days_ago = datetime.utcnow() - timedelta(days=90) recent_reports = await db["reports"].find( { "created_at": {"$gte": ninety_days_ago}, "status": "completed" }, {"website": 1, "user_id": 1} ).to_list(length=None) for report in recent_reports: website = report.get("website", "").rstrip("/") websites_to_check.add(website) if website not in user_website_map: # Fetch user info if not already loaded user = await db["users"].find_one( {"_id": report["user_id"]}, {"_id": 1, "email": 1, "username": 1} ) if user: user_website_map[website] = [user] else: # Check if this user is already in the map user_id_str = str(report["user_id"]) user_ids = [str(u["_id"]) for u in user_website_map[website]] if user_id_str not in user_ids: user = await db["users"].find_one( {"_id": report["user_id"]}, {"_id": 1, "email": 1, "username": 1} ) if user: user_website_map[website].append(user) self.last_check_stats["total_websites"] = len(websites_to_check) print(f"Total websites to check: {len(websites_to_check)}") # Step 3: Check SSL for each website for website in websites_to_check: await self.check_website_ssl(website, user_website_map.get(website, [])) # Print summary print(f"\nSSL Check Complete:") print(f" Total: {self.last_check_stats['total_websites']}") print(f" Successful: {self.last_check_stats['successful_checks']}") print(f" Failed: {self.last_check_stats['failed_checks']}") print(f" Alerts Sent: {self.last_check_stats['alerts_sent']}") except Exception as e: print(f"Error in daily SSL check: {str(e)}") import traceback traceback.print_exc() async def check_website_ssl(self, website_url: str, users: List[Dict]): """ Check SSL certificate for a single website and update database. Args: website_url: The website URL to check users: List of user documents who own this website """ if not users: print(f"Skipping {website_url} - no associated users") return try: # Check SSL certificate ssl_info = await check_ssl_certificate_async(website_url) if ssl_info.get("error"): self.last_check_stats["failed_checks"] += 1 print(f"Failed to check SSL for {website_url}: {ssl_info.get('error')}") else: self.last_check_stats["successful_checks"] += 1 print(f"SSL checked for {website_url}: " f"{'Valid' if ssl_info.get('ssl_valid') else 'Invalid'}, " f"{ssl_info.get('days_until_expiry', 'N/A')} days remaining") # Step 4: Update database with SSL status await self.update_ssl_status_in_db(website_url, ssl_info, users) # Step 5: Send email alerts if needed await self.send_ssl_alerts_if_needed(website_url, ssl_info, users) except Exception as e: print(f"Error checking SSL for {website_url}: {str(e)}") self.last_check_stats["failed_checks"] += 1 async def update_ssl_status_in_db( self, website_url: str, ssl_info: Dict[str, Any], users: List[Dict] ): """ Update SSL status in MongoDB for all reports related to this website. Args: website_url: The website URL ssl_info: SSL certificate information users: List of users associated with this website """ try: # Update the most recent report for this website latest_report = await db["reports"].find_one( {"website": website_url}, sort=[("created_at", -1)] ) if latest_report: ssl_status = { "has_ssl": ssl_info.get("has_ssl", False), "ssl_valid": ssl_info.get("ssl_valid", False), "is_expired": ssl_info.get("is_expired", False), "days_until_expiry": ssl_info.get("days_until_expiry"), "expiry_date": ssl_info.get("expiry_date"), "issuer": ssl_info.get("issuer"), "last_checked": datetime.utcnow() } await db["reports"].update_one( {"_id": latest_report["_id"]}, {"$set": {"ssl_status": ssl_status}} ) print(f"Updated SSL status in DB for {website_url}") except Exception as e: print(f"Error updating SSL status in DB for {website_url}: {str(e)}") async def send_ssl_alerts_if_needed( self, website_url: str, ssl_info: Dict[str, Any], users: List[Dict] ): """ Send SSL alert emails to users if certificate is expired or expiring soon. Prevents duplicate alerts by checking if an alert was sent in the last 24 hours. Args: website_url: The website URL ssl_info: SSL certificate information users: List of users to notify """ try: if not ssl_info.get("has_ssl"): # No SSL certificate - don't send alerts for this case return is_expired = ssl_info.get("is_expired", False) days_until_expiry = ssl_info.get("days_until_expiry") # Determine if we should send an alert should_send_alert = False alert_type = None if is_expired: should_send_alert = True alert_type = "expired" elif days_until_expiry is not None and days_until_expiry <= SSL_URGENT_THRESHOLD_DAYS: should_send_alert = True alert_type = "urgent" # 3 days or less elif days_until_expiry is not None and days_until_expiry <= SSL_ALERT_THRESHOLD_DAYS: should_send_alert = True alert_type = "warning" # 4-7 days if not should_send_alert: return # Check if alert was already sent in the last 24 hours to prevent spam twenty_four_hours_ago = datetime.utcnow() - timedelta(hours=24) for user in users: user_email = user.get("email") username = user.get("username") if not user_email: continue # Check last alert time from database recent_alert = await db["reports"].find_one( { "website": website_url, "user_id": user["_id"], "ssl_alert_sent": True, "ssl_alert_sent_at": {"$gte": twenty_four_hours_ago} } ) if recent_alert: print(f"Skipping alert for {user_email} - already sent in last 24h") continue # Get backend URL for email template backend_url = os.getenv("BACKEND_URL", "http://localhost:8000") # Send the alert email ssl_alert_result = send_ssl_alert_email( to_email=user_email, website_url=website_url, username=username, ssl_info=ssl_info, backend_url=backend_url ) if ssl_alert_result["success"]: self.last_check_stats["alerts_sent"] += 1 # Update database with alert status await db["reports"].update_one( { "website": website_url, "user_id": user["_id"] }, { "$set": { "ssl_alert_sent": True, "ssl_alert_sent_at": datetime.utcnow(), "ssl_status": ssl_info } }, upsert=False ) alert_prefix = "[EXPIRED]" if alert_type == "expired" else "[URGENT]" if alert_type == "urgent" else "[WARNING]" print(f"{alert_prefix} Alert sent to {user_email} for {website_url} ({alert_type})") else: print(f"Failed to send alert to {user_email}: {ssl_alert_result.get('error')}") except Exception as e: print(f"Error sending SSL alerts for {website_url}: {str(e)}") def get_stats(self) -> Dict[str, Any]: """ Get current SSL monitor statistics. Returns: dict: Monitor statistics """ return { **self.last_check_stats, "is_running": self.is_running, "next_run": self.scheduler.get_job("daily_ssl_check").next_run_time.isoformat() if self.scheduler.get_job("daily_ssl_check") else None } # Global instance ssl_monitor = SSLMonitor() def get_ssl_monitor() -> SSLMonitor: """Get the global SSL monitor instance.""" return ssl_monitor def start_ssl_monitor(): """ Start the SSL monitoring service. Call this when the FastAPI application starts. """ ssl_monitor.start() return ssl_monitor def stop_ssl_monitor(): """ Stop the SSL monitoring service. Call this when the FastAPI application shuts down. """ ssl_monitor.stop()