import psutil import os import requests import smtplib import distro import subprocess import time from email.mime.text import MIMEText try: from .validation import is_safe_external_url except ImportError: from validation import is_safe_external_url # Global state for network speed calculation _net_state = { 'last_bytes_sent': 0, 'last_bytes_recv': 0, 'last_time': 0, 'speed_sent_bps': 0, 'speed_recv_bps': 0 } def get_system_stats(): """Collects system statistics.""" cpu_percent = psutil.cpu_percent(interval=1) mem = psutil.virtual_memory() ram_percent = mem.percent disk = psutil.disk_usage('/') disk_percent = disk.percent load = os.getloadavg() raid_status = check_raid() os_info = f"{distro.name()} {distro.version()} ({distro.codename().capitalize()})" # Network Traffic net = psutil.net_io_counters() # net.bytes_sent, net.bytes_recv # Network Speed Calculation global _net_state current_time = time.time() if _net_state['last_time'] == 0: # Init _net_state['last_bytes_sent'] = net.bytes_sent _net_state['last_bytes_recv'] = net.bytes_recv _net_state['last_time'] = current_time speed_sent = 0 speed_recv = 0 else: # Calculate delta_time = current_time - _net_state['last_time'] if delta_time > 0: # Avoid division by zero delta_bytes_sent = net.bytes_sent - _net_state['last_bytes_sent'] delta_bytes_recv = net.bytes_recv - _net_state['last_bytes_recv'] speed_sent = delta_bytes_sent / delta_time speed_recv = delta_bytes_recv / delta_time # Update _net_state['last_bytes_sent'] = net.bytes_sent _net_state['last_bytes_recv'] = net.bytes_recv _net_state['last_time'] = current_time _net_state['speed_sent_bps'] = speed_sent _net_state['speed_recv_bps'] = speed_recv else: speed_sent = _net_state['speed_sent_bps'] speed_recv = _net_state['speed_recv_bps'] return { 'cpu': cpu_percent, 'ram': ram_percent, 'ram_total': mem.total, 'ram_used': mem.used, 'disk': disk_percent, 'disk_total': disk.total, 'disk_used': disk.used, 'net_sent': net.bytes_sent, 'net_recv': net.bytes_recv, 'speed_sent': speed_sent, 'speed_recv': speed_recv, 'load': load, 'raid': raid_status, 'os_info': os_info } def check_raid(): """Checks software RAID status from /proc/mdstat.""" if not os.path.exists('/proc/mdstat'): return "No RAID detected" try: with open('/proc/mdstat', 'r') as f: content = f.read() if 'blocks' not in content: return "Inactive" if '_' in content: return "DEGRADED" return "OK" except Exception as e: return f"Error: {str(e)}" def send_alert(settings, subject, message, target_email=None, disable_email=False): """Sends alert via SMTP, Webhook, and saves to DB.""" try: from .database import db from .models import Alert except ImportError: from database import db from models import Alert import json if settings is None: return # Save to DB try: # Determine type based on subject (simple heuristic) alert_type = "System" if "Quota" in subject or "Disk" in subject: alert_type = "Quota" elif "Bandwidth" in subject: alert_type = "Bandwidth" elif "Security" in subject: alert_type = "Security" new_alert = Alert(type=alert_type, subject=subject, message=message) db.session.add(new_alert) db.session.commit() except Exception as e: print(f"Failed to save alert to DB: {e}") # Extract settings for the thread webhook_url = settings.webhook_url webhook_headers_str = settings.webhook_headers smtp_server = settings.smtp_server smtp_port = settings.smtp_port smtp_user = settings.smtp_user smtp_pass = settings.smtp_pass alert_email = settings.alert_email def _send_async(): import requests from email.mime.text import MIMEText import smtplib import json if webhook_url and is_safe_external_url(webhook_url, block_internal=True): try: headers = {} if webhook_headers_str: try: headers = json.loads(webhook_headers_str) except: print("Invalid Webhook Headers JSON") requests.post(webhook_url, json={'subject': subject, 'message': message}, headers=headers, timeout=5) except Exception as e: print(f"Webhook failed: {e}") if not disable_email and smtp_server and (alert_email or target_email): try: msg = MIMEText(message) msg['Subject'] = f"[Limristem Web Alert] {subject}" msg['From'] = smtp_user msg['To'] = target_email if target_email else alert_email if msg['To']: with smtplib.SMTP(smtp_server, smtp_port) as server: server.starttls() server.login(smtp_user, smtp_pass) server.send_message(msg) except Exception as e: print(f"SMTP failed: {e}") import threading threading.Thread(target=_send_async, daemon=True).start() def check_thresholds_and_alert(settings): """Called periodically to check thresholds.""" stats = get_system_stats() alerts = [] if stats['cpu'] > settings.cpu_limit: alerts.append(f"High CPU Load: {stats['cpu']}%") if stats['ram'] > settings.ram_limit: alerts.append(f"High RAM Usage: {stats['ram']}%") if stats['disk'] > settings.disk_limit: alerts.append(f"Low Disk Space: {stats['disk']}% used") if stats['raid'] == "DEGRADED": alerts.append("RAID Array is DEGRADED!") if alerts: msg_body = "\n".join(alerts) send_alert(settings, "System Critical Warning", msg_body) def update_traffic_stats(db_session, HostModel): """Parses Nginx logs to update traffic stats, tracking file offsets.""" hosts = HostModel.query.all() for host in hosts: log_path = f"/var/log/nginx/{host.domain}.access.log" if os.path.exists(log_path): try: stat = os.stat(log_path) current_inode = stat.st_ino current_size = stat.st_size # Check for rotation if host.last_log_inode != 0 and host.last_log_inode != current_inode: # Log rotated. Reset offset to 0. host.last_log_offset = 0 host.last_log_inode = current_inode # If file shrank (truncated), reset offset if current_size < host.last_log_offset: host.last_log_offset = 0 # Read new data if current_size > host.last_log_offset: # Revert to Python reading for maximum reliability as requested # Reading chunk by chunk to avoid memory issues try: with open(log_path, 'rb') as f: f.seek(host.last_log_offset) new_bytes_traffic = 0 # Read line by line from current position for line in f: try: # Decode safely line_str = line.decode('utf-8', errors='ignore') parts = line_str.split() # Combined Log Format: IP - - [Date] "Req" Status Bytes ... # Index 9 usually. if len(parts) > 9: size_str = parts[9] if size_str != '-': new_bytes_traffic += int(size_str) except: pass if host.current_traffic_bytes is None: host.current_traffic_bytes = 0 host.current_traffic_bytes += new_bytes_traffic host.last_log_offset = f.tell() # print(f"[Debug] Traffic {host.domain}: +{new_bytes_traffic} bytes") except Exception as ex: print(f"Log read error for {host.domain}: {ex}") except Exception as e: print(f"Error accessing log for {host.domain}: {e}") db_session.commit() def check_quotas_and_lock(db_session, SystemUserModel, HostModel, settings): """Checks disk usage, bandwidth, and resources against limits and locks/suspends if exceeded.""" import subprocess from datetime import datetime, timedelta try: from . import user_host_manager except ImportError: import user_host_manager # --- Check User Disk Quotas & Resources --- users = SystemUserModel.query.all() for user in users: # 1. Disk Quota used_bytes = 0 try: if os.path.exists(user.home_dir): res = subprocess.run(['du', '-sb', user.home_dir], capture_output=True, text=True) if res.returncode == 0: used_bytes = int(res.stdout.split()[0]) user.current_disk_usage = used_bytes if user.quota_limit_mb > 0: limit_bytes = int(user.quota_limit_mb) * 1024 * 1024 if used_bytes > limit_bytes: print(f"[Monitor] User {user.username} over quota: {used_bytes} > {limit_bytes}") if not user.is_suspended: # Lock User System Account subprocess.run(['usermod', '-L', user.username], check=False) user.is_suspended = True user.suspension_reason = "disk_full_user" send_alert(settings, "User Quota Exceeded", f"User {user.username} exceeded quota. Locked.") # Notify User if user.email: send_alert(settings, "Account Suspended: Disk Quota Exceeded", f"Dear {user.username},\nYour account has exceeded its disk quota of {user.quota_limit_mb}MB.\nPlease contact support.", target_email=user.email) else: # Auto unlock if below quota? For MVP, only if suspended by us for this reason if user.is_suspended and user.suspension_reason == "disk_full_user": subprocess.run(['usermod', '-U', user.username], check=False) user.is_suspended = False user.suspension_reason = None except Exception as e: print(f"User check error {user.username}: {e}") # 2. Resource Limits (CPU/RAM/IO) - Placeholder Logic # Real enforcement requires monitoring process tree or cgroups. # Here we simulate the logic structure requested. # We need to find PIDs owned by user. # psutil.process_iter(['username', 'cpu_percent', 'memory_info']) # Sum up usage. if user.cpu_limit_percent > 0 or user.ram_limit_mb > 0: try: total_cpu = 0.0 total_ram_mb = 0.0 for proc in psutil.process_iter(['username', 'cpu_percent', 'memory_info']): if proc.info['username'] == user.username: total_cpu += proc.info['cpu_percent'] total_ram_mb += proc.info['memory_info'].rss / 1024 / 1024 # Check CPU if user.cpu_limit_percent > 0: if total_cpu > (user.cpu_limit_percent * 1.2): # 120% # Permanent Suspend Logic pass # TODO: Implement permanent suspend elif total_cpu > (user.cpu_limit_percent * 1.1): # 110% # Temporary Suspend (30m) pass elif total_cpu > user.cpu_limit_percent: # 100% # Notify pass except: pass # --- Check Host Quotas (Disk & Traffic) --- hosts = HostModel.query.all() for host in hosts: # Fetch user for home_dir access needed for unsuspension user = SystemUserModel.query.get(host.user_id) # 1. Disk Quota used_bytes = 0 try: if os.path.exists(host.root_dir): res = subprocess.run(['du', '-sb', host.root_dir], capture_output=True, text=True) if res.returncode == 0: used_bytes = int(res.stdout.split()[0]) host.current_disk_usage = used_bytes if host.quota_limit_mb > 0: limit_bytes = int(host.quota_limit_mb) * 1024 * 1024 if used_bytes > limit_bytes: print(f"[Monitor] Host {host.domain} over disk quota: {used_bytes} > {limit_bytes}") if not host.is_suspended: user_host_manager.suspend_host(host.domain, "disk_full") host.is_suspended = True host.suspension_reason = "disk_full" send_alert(settings, "Host Suspended: Disk Full", f"Host {host.domain} exceeded disk quota.") else: # Unsuspend if condition cleared if host.is_suspended and host.suspension_reason == "disk_full": print(f"[Monitor] Un-suspending {host.domain} (Disk OK)") user_host_manager.unsuspend_host( host.domain, host.php_version, user.home_dir, host.speed_limit_kbps, conn_limit=host.max_concurrent_connections, max_body_size_mb=host.nginx_max_body_size_mb, client_body_timeout=host.nginx_client_body_timeout, client_header_timeout=host.nginx_client_header_timeout, keepalive_timeout=host.nginx_keepalive_timeout ) host.is_suspended = False host.suspension_reason = None except Exception as e: print(f"Host disk check error {host.domain}: {e}") # 2. Traffic Quota if host.traffic_limit_mb > 0: traffic_mb = (host.current_traffic_bytes or 0) / 1024 / 1024 if traffic_mb > host.traffic_limit_mb: print(f"[Monitor] Host {host.domain} over bandwidth: {traffic_mb}MB > {host.traffic_limit_mb}MB") if not host.is_suspended: user_host_manager.suspend_host(host.domain, "bandwidth") host.is_suspended = True host.suspension_reason = "bandwidth" send_alert(settings, "Host Suspended: Bandwidth", f"Host {host.domain} exceeded traffic limit.") else: if host.is_suspended and host.suspension_reason == "bandwidth": print(f"[Monitor] Un-suspending {host.domain} (Bandwidth OK)") user_host_manager.unsuspend_host( host.domain, host.php_version, user.home_dir, host.speed_limit_kbps, conn_limit=host.max_concurrent_connections, max_body_size_mb=host.nginx_max_body_size_mb, client_body_timeout=host.nginx_client_body_timeout, client_header_timeout=host.nginx_client_header_timeout, keepalive_timeout=host.nginx_keepalive_timeout ) host.is_suspended = False host.suspension_reason = None db_session.commit()