import os import sys import json import click import secrets import ipaddress import uuid from datetime import datetime, timedelta from flask import Flask, render_template, request, redirect, url_for, flash, session, send_from_directory, jsonify, g, send_file from werkzeug.security import generate_password_hash, check_password_hash from werkzeug.utils import secure_filename from werkzeug.middleware.proxy_fix import ProxyFix from apscheduler.schedulers.background import BackgroundScheduler from flask_wtf.csrf import CSRFProtect, CSRFError try: from .database import db from .models import Admin, Settings, Host, SystemUser, BackupJob, LoginAttempt, BannedIP, Alert from . import monitor, php_manager, user_host_manager, backup_manager, ssl_manager, mysql_manager, task_manager, update_manager from .validation import ( is_safe_external_url, is_valid_db_identifier, is_valid_domain_name, is_valid_hostname, is_valid_origin_host, is_valid_system_username, is_valid_php_pm, is_valid_php_size, ) from .version import VERSION except ImportError: from database import db from models import Admin, Settings, Host, SystemUser, BackupJob, LoginAttempt, BannedIP, Alert import monitor import php_manager import user_host_manager import backup_manager import ssl_manager import mysql_manager import task_manager import update_manager from validation import ( is_safe_external_url, is_valid_db_identifier, is_valid_domain_name, is_valid_hostname, is_valid_origin_host, is_valid_system_username, ) from version import VERSION # Limristem Web # Un software di Limristem distribuito in licenza CC BY-NC 4.0 scheduler_instance = None INSTALL_DIR = '/opt/limristem-web' ALLOWED_LOG_DIRS = ('/var/log/nginx', '/var/log') ALLOWED_UPLOAD_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'svg', 'webp', 'ico'} def _config_path(instance_path): return os.path.join(instance_path, 'config.json') def _load_or_create_secret(instance_path): config_path = _config_path(instance_path) secret_key = None if os.path.exists(config_path): try: with open(config_path, 'r') as f: conf = json.load(f) candidate = conf.get('SECRET_KEY') if candidate and candidate != 'dev': return candidate except Exception: pass secret_key = secrets.token_hex(32) os.makedirs(instance_path, exist_ok=True) with open(config_path, 'w') as f: json.dump({'SECRET_KEY': secret_key}, f) try: os.chmod(config_path, 0o600) except OSError: pass return secret_key def _parse_non_negative_int(value, default=0, minimum=0, maximum=None): try: parsed = int(value) except (TypeError, ValueError): return default if parsed < minimum: return default if maximum is not None and parsed > maximum: return default return parsed def _allowed_upload(filename): if not filename or '.' not in filename: return False ext = filename.rsplit('.', 1)[1].lower() return ext in ALLOWED_UPLOAD_EXTENSIONS def _save_uploaded_asset(file_storage, upload_folder, prefix, max_size_kb=None): filename = secure_filename(file_storage.filename or '') if not _allowed_upload(filename): raise ValueError('Unsupported file type') if max_size_kb is not None: file_storage.seek(0, os.SEEK_END) size_bytes = file_storage.tell() file_storage.seek(0) if size_bytes > max_size_kb * 1024: raise ValueError(f'The file must be at most {max_size_kb}KB') safe_name = f"{prefix}_{int(datetime.now().timestamp())}_{filename}" path = os.path.join(upload_folder, safe_name) file_storage.save(path) try: os.chmod(path, 0o644) except OSError: pass return safe_name def _is_allowed_log_path(app, path): if not os.path.isabs(path): safe_name = secure_filename(path) if not safe_name: return None return os.path.join(app.instance_path, 'logs', safe_name) real_path = os.path.realpath(os.path.normpath(path)) if real_path.startswith(os.path.realpath(os.path.join(app.instance_path, 'logs')) + os.sep): return real_path allowed = set() for host in Host.query.all(): allowed.add(os.path.realpath(f"/var/log/nginx/{host.domain}.access.log")) allowed.add(os.path.realpath(f"/var/log/nginx/{host.domain}.error.log")) for ver in php_manager.get_installed_php_versions(): for candidate in ( f"/var/log/php{ver}-fpm.log", f"/var/log/php{ver}-fpm/error.log", ): allowed.add(os.path.realpath(candidate)) for host in Host.query.filter_by(php_version=ver).all(): for candidate in ( f"/var/log/php{ver}-fpm/{host.domain}.error.log", f"/var/log/php{ver}-fpm/{host.domain}-error.log", ): allowed.add(os.path.realpath(candidate)) return real_path if real_path in allowed else None def _trusted_origins_for_request(): if not request.host or not is_valid_origin_host(request.host): return [] origin = request.host_url.rstrip('/') return [origin] def _get_mysql_backup_config(root_path=None): """Load MySQL connection config from the stored config file.""" paths = [] if root_path: paths.append(os.path.join(root_path, 'instance/mysql_config.json')) paths.append('/opt/limristem-web/instance/mysql_config.json') for path in paths: if os.path.exists(path): with open(path) as f: return json.load(f) return {} def _validate_mysql_backup_settings(mysql_enabled, mysql_db_mode, mysql_db_name, mysql_db_list): """Validate MySQL backup form fields. Returns an error message string or None.""" if not mysql_enabled: return None if mysql_db_mode not in ('all', 'single', 'list'): return "Invalid MySQL db mode: must be 'all', 'single', or 'list'" if mysql_db_mode == 'single' and not mysql_db_name: return "MySQL db mode 'single' requires a database name" if mysql_db_mode == 'single' and not is_valid_db_identifier(mysql_db_name.strip()): return "Invalid MySQL database name" if mysql_db_mode == 'list' and not mysql_db_list: return "MySQL db mode 'list' requires at least one database name" if mysql_db_mode == 'list': invalid = [db for db in (name.strip() for name in mysql_db_list.split(',')) if db and not is_valid_db_identifier(db)] if invalid: return f"Invalid MySQL database name(s): {', '.join(invalid)}" return None def create_app(test_config=None): global scheduler_instance app = Flask(__name__, instance_relative_config=True) # Apply ProxyFix to handle X-Forwarded-Prefix for correct URL generation behind Nginx proxy # NOTE: x_host is enabled, meaning request.host respects X-Forwarded-Host if present. # If accessing via port 4316 directly via Nginx, Nginx sets Host header. # If accessing via IP, Host header is IP:Port. app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1) try: os.makedirs(app.instance_path, exist_ok=True) except OSError: pass secret_key = _load_or_create_secret(app.instance_path) app.config.from_mapping( SECRET_KEY=secret_key, SQLALCHEMY_DATABASE_URI='sqlite:///' + os.path.join(app.instance_path, 'limristem_web.sqlite'), WTF_CSRF_SSL_STRICT=False, SESSION_COOKIE_SECURE=True, SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE='Lax', MAX_CONTENT_LENGTH=100 * 1024 * 1024, ) if test_config: app.config.from_mapping(test_config) # Configure Uploads UPLOAD_FOLDER = os.path.join(app.instance_path, 'uploads') if not os.path.exists(UPLOAD_FOLDER): os.makedirs(UPLOAD_FOLDER, mode=0o700) app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER # Initialize CSRF Protection csrf = CSRFProtect(app) # Return JSON for CSRF errors on JSON/XHR API endpoints @app.errorhandler(CSRFError) def handle_csrf_error(e): if request.is_json or 'X-CSRFToken' in request.headers: return jsonify({'error': f'CSRF validation failed: {e.description}'}), 400 return render_template('login.html', error=e.description), 400 from flask_babel import Babel def get_locale(): from flask import g, request, session # If user is logged in and has a language preference if getattr(g, 'user', None) and g.user.language: return g.user.language # Fallback to browser's best match return request.accept_languages.best_match(['en', 'it', 'fr', 'es', 'de']) babel = Babel(app, locale_selector=get_locale) @app.after_request def add_security_headers(response): response.headers['X-Frame-Options'] = 'DENY' response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' response.headers['Permissions-Policy'] = 'geolocation=(), microphone=(), camera=()' response.headers['Content-Security-Policy'] = ( "default-src 'self'; " "img-src 'self' data: https:; " "style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net https://cdnjs.cloudflare.com https://fonts.googleapis.com; " "font-src 'self' data: https://cdnjs.cloudflare.com https://fonts.gstatic.com; " "script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net https://cdnjs.cloudflare.com; " "connect-src 'self'; base-uri 'self'; form-action 'self'; frame-ancestors 'none'" ) return response # Register custom Jinja2 filter for JSON parsing in templates @app.template_filter('from_json') def from_json_filter(value): try: return json.loads(value) if value else {} except Exception: return {} @app.template_filter('format_time') def format_time_filter(dt, format='%Y-%m-%d %H:%M'): if dt is None: return "" from flask import g import pytz tz_name = getattr(g.user, 'timezone', 'UTC') if getattr(g, 'user', None) else 'UTC' try: tz = pytz.timezone(tz_name) except: tz = pytz.UTC if dt.tzinfo is None: dt = pytz.utc.localize(dt) return dt.astimezone(tz).strftime(format) db.init_app(app) with app.app_context(): # Suppress Paramiko Cryptography Warnings import warnings try: from cryptography.utils import CryptographyDeprecationWarning warnings.filterwarnings("ignore", category=CryptographyDeprecationWarning) except ImportError: pass # Robust DB Creation try: db.create_all() except Exception as e: # If tables exist, ignore (OperationalError) if "already exists" not in str(e): print(f"DB Create Error (ignored): {e}") # Migration Helper: Check for new columns and add them if missing # Moved up to ensure schema is correct before querying models from sqlalchemy import text try: with db.engine.connect() as conn: # Check Settings Customization try: conn.execute(text("SELECT app_name FROM settings LIMIT 1")) except: conn.execute(text("ALTER TABLE settings ADD COLUMN app_name VARCHAR(100) DEFAULT 'Limristem Web'")) conn.execute(text("ALTER TABLE settings ADD COLUMN logo_url VARCHAR(255)")) conn.execute(text("ALTER TABLE settings ADD COLUMN favicon_url VARCHAR(255)")) conn.execute(text("ALTER TABLE settings ADD COLUMN custom_message TEXT")) conn.commit() # Check Host Connection Limit try: conn.execute(text("SELECT max_concurrent_connections FROM host LIMIT 1")) except: conn.execute(text("ALTER TABLE host ADD COLUMN max_concurrent_connections INTEGER DEFAULT 10")) conn.commit() # Check Auto Update try: conn.execute(text("SELECT auto_update FROM settings LIMIT 1")) except: conn.execute(text("ALTER TABLE settings ADD COLUMN auto_update BOOLEAN DEFAULT 0")) conn.execute(text("ALTER TABLE settings ADD COLUMN last_update_check DATETIME")) conn.commit() # Check Current Version try: conn.execute(text("SELECT current_version FROM settings LIMIT 1")) except: conn.execute(text("ALTER TABLE settings ADD COLUMN current_version VARCHAR(20)")) conn.commit() # Check Server Hostname & Panel SSL (NEW MIGRATION) try: conn.execute(text("SELECT server_hostname FROM settings LIMIT 1")) except: conn.execute(text("ALTER TABLE settings ADD COLUMN server_hostname VARCHAR(200)")) conn.execute(text("ALTER TABLE settings ADD COLUMN panel_ssl_type VARCHAR(50)")) conn.commit() # Check Admin Profile Fields (NEW MIGRATION) # Apply column individually to ensure robustness against partial migrations columns_to_add = [ "avatar_url VARCHAR(255)", "nickname VARCHAR(100)", "timezone VARCHAR(50) DEFAULT 'UTC'", "language VARCHAR(10) DEFAULT 'en'", "totp_secret VARCHAR(32)", "totp_enabled BOOLEAN DEFAULT 0", "email VARCHAR(120)", "notify_on_login BOOLEAN DEFAULT 0", "notify_on_failed_login BOOLEAN DEFAULT 0", "notify_update_available BOOLEAN DEFAULT 1", "notify_update_installed BOOLEAN DEFAULT 1" ] for col_def in columns_to_add: col_name = col_def.split(" ")[0] try: conn.execute(text(f"SELECT {col_name} FROM admin LIMIT 1")) except: try: conn.execute(text(f"ALTER TABLE admin ADD COLUMN {col_def}")) except Exception as e: print(f"Failed to add column {col_name}: {e}") try: conn.commit() except: pass # Check Alert Table try: conn.execute(text("SELECT id FROM alert LIMIT 1")) except: conn.execute(text(""" CREATE TABLE alert ( id INTEGER PRIMARY KEY, type VARCHAR(50), subject VARCHAR(200), message TEXT, timestamp DATETIME, is_read BOOLEAN ) """)) conn.commit() # Check SystemUser # Logic: Check if sftp_user exists, if so, rename it to system_user? # Or just assume fresh install for the rename since we are dev. # But for migration safety, let's check. try: conn.execute(text("SELECT id FROM sftp_user LIMIT 1")) # If success, rename table conn.execute(text("ALTER TABLE sftp_user RENAME TO system_user")) conn.commit() except: pass try: conn.execute(text("SELECT current_disk_usage FROM system_user LIMIT 1")) except: conn.execute(text("ALTER TABLE system_user ADD COLUMN current_disk_usage BIGINT DEFAULT 0")) conn.commit() try: conn.execute(text("SELECT email FROM system_user LIMIT 1")) except: conn.execute(text("ALTER TABLE system_user ADD COLUMN email VARCHAR(120)")) conn.execute(text("ALTER TABLE system_user ADD COLUMN cpu_limit_percent INTEGER DEFAULT 0")) conn.execute(text("ALTER TABLE system_user ADD COLUMN ram_limit_mb INTEGER DEFAULT 0")) conn.execute(text("ALTER TABLE system_user ADD COLUMN io_limit_mb_s INTEGER DEFAULT 0")) conn.execute(text("ALTER TABLE system_user ADD COLUMN is_suspended BOOLEAN DEFAULT 0")) conn.execute(text("ALTER TABLE system_user ADD COLUMN suspension_reason VARCHAR(50)")) conn.execute(text("ALTER TABLE system_user ADD COLUMN suspension_end_time DATETIME")) conn.execute(text("ALTER TABLE system_user ADD COLUMN violation_count INTEGER DEFAULT 0")) conn.execute(text("ALTER TABLE system_user ADD COLUMN last_violation_time DATETIME")) conn.commit() # Check Host try: conn.execute(text("SELECT current_disk_usage FROM host LIMIT 1")) except: conn.execute(text("ALTER TABLE host ADD COLUMN current_disk_usage BIGINT DEFAULT 0")) conn.commit() try: conn.execute(text("SELECT is_suspended FROM host LIMIT 1")) except: conn.execute(text("ALTER TABLE host ADD COLUMN is_suspended BOOLEAN DEFAULT 0")) conn.execute(text("ALTER TABLE host ADD COLUMN suspension_reason VARCHAR(50)")) conn.commit() # Check PHP Settings columns try: conn.execute(text("SELECT php_pm FROM host LIMIT 1")) except: conn.execute(text("ALTER TABLE host ADD COLUMN php_pm VARCHAR(20) DEFAULT 'ondemand'")) conn.execute(text("ALTER TABLE host ADD COLUMN php_max_children INTEGER DEFAULT 5")) conn.execute(text("ALTER TABLE host ADD COLUMN php_start_servers INTEGER DEFAULT 2")) conn.execute(text("ALTER TABLE host ADD COLUMN php_min_spare_servers INTEGER DEFAULT 1")) conn.execute(text("ALTER TABLE host ADD COLUMN php_max_spare_servers INTEGER DEFAULT 3")) conn.execute(text("ALTER TABLE host ADD COLUMN php_memory_limit VARCHAR(20) DEFAULT '128M'")) conn.execute(text("ALTER TABLE host ADD COLUMN php_upload_max_filesize VARCHAR(20) DEFAULT '10M'")) conn.execute(text("ALTER TABLE host ADD COLUMN php_post_max_size VARCHAR(20) DEFAULT '10M'")) conn.execute(text("ALTER TABLE host ADD COLUMN php_max_execution_time INTEGER DEFAULT 30")) conn.commit() # Check Backup Retention try: conn.execute(text("SELECT retention_days FROM backup_job LIMIT 1")) except: conn.execute(text("ALTER TABLE backup_job ADD COLUMN retention_days INTEGER DEFAULT 30")) conn.execute(text("ALTER TABLE backup_job ADD COLUMN retention_count INTEGER DEFAULT 10")) conn.commit() # Check Backup MySQL columns try: conn.execute(text("SELECT mysql_enabled FROM backup_job LIMIT 1")) except: conn.execute(text("ALTER TABLE backup_job ADD COLUMN mysql_enabled BOOLEAN DEFAULT 0")) conn.execute(text("ALTER TABLE backup_job ADD COLUMN mysql_db_mode VARCHAR(10) DEFAULT 'all'")) conn.execute(text("ALTER TABLE backup_job ADD COLUMN mysql_db_name VARCHAR(100)")) conn.execute(text("ALTER TABLE backup_job ADD COLUMN mysql_db_list TEXT")) conn.commit() # Check Panel Limits & Timeout try: conn.execute(text("SELECT panel_upload_limit_mb FROM settings LIMIT 1")) except: conn.execute(text("ALTER TABLE settings ADD COLUMN panel_upload_limit_mb INTEGER DEFAULT 100")) conn.commit() try: conn.execute(text("SELECT panel_timeout_seconds FROM settings LIMIT 1")) except: conn.execute(text("ALTER TABLE settings ADD COLUMN panel_timeout_seconds INTEGER DEFAULT 300")) conn.commit() # Check Host Type columns try: conn.execute(text("SELECT host_type FROM host LIMIT 1")) except: conn.execute(text("ALTER TABLE host ADD COLUMN host_type VARCHAR(20) DEFAULT 'php'")) conn.execute(text("ALTER TABLE host ADD COLUMN template VARCHAR(50) DEFAULT 'default'")) conn.commit() # Check is_default column try: conn.execute(text("SELECT is_default FROM host LIMIT 1")) except: conn.execute(text("ALTER TABLE host ADD COLUMN is_default BOOLEAN DEFAULT 0")) conn.commit() # Check Nginx Timeout & Payload Settings (NEW MIGRATION) try: conn.execute(text("SELECT nginx_max_body_size_mb FROM host LIMIT 1")) except: conn.execute(text("ALTER TABLE host ADD COLUMN nginx_max_body_size_mb INTEGER DEFAULT 0")) conn.execute(text("ALTER TABLE host ADD COLUMN nginx_client_body_timeout INTEGER DEFAULT 0")) conn.execute(text("ALTER TABLE host ADD COLUMN nginx_client_header_timeout INTEGER DEFAULT 0")) conn.execute(text("ALTER TABLE host ADD COLUMN nginx_keepalive_timeout INTEGER DEFAULT 0")) conn.commit() except Exception as e: print(f"Migration check failed (safe to ignore if fresh install): {e}") # Ensure default admin exists only if no admin exists # Note: Installer script handles setting custom credentials. # This block is fail-safe for manual runs. try: if not Admin.query.first(): random_pass = secrets.token_hex(16) db.session.add(Admin(username='admin', password=generate_password_hash(random_pass))) db.session.commit() print(f"WARNING: Created default admin 'admin' with a random password. Use 'flask show-admin --reveal' to reset it.") except Exception: db.session.rollback() pass # Race condition or already exists settings = Settings.query.first() if not settings: settings = Settings() settings.current_version = VERSION db.session.add(settings) db.session.commit() else: if settings.current_version != VERSION: old_ver = settings.current_version or "unknown" # Create one global alert and send emails to admins who want it msg = f"Limristem Web has been successfully updated from {old_ver} to {VERSION}." admins = Admin.query.all() emails_to_notify = [a.email for a in admins if a.notify_update_installed and a.email] # Always create the DB alert, but only send email if at least one admin wants it if emails_to_notify: for email in emails_to_notify: monitor.send_alert(settings, "Update Installed", msg, target_email=email) else: # No emails wanted, just create DB alert monitor.send_alert(settings, "Update Installed", msg, disable_email=True) settings.current_version = VERSION db.session.commit() # Update Helper def run_update_logic(settings): has_update, new_version, err = update_manager.check_for_updates() if has_update: msg = f"New version available: {new_version}. Current: {VERSION}" existing_alert = Alert.query.filter(Alert.subject == "Update Available", Alert.message.like(f"%{new_version}%")).first() if not existing_alert: admins = Admin.query.all() emails_to_notify = [a.email for a in admins if a.notify_update_available and a.email] if emails_to_notify: for email in emails_to_notify: monitor.send_alert(settings, "Update Available", msg, target_email=email) else: monitor.send_alert(settings, "Update Available", msg, disable_email=True) if settings.auto_update: success, u_msg = update_manager.perform_update() if success: # The restart logic will catch the successful install and notify admins pass else: monitor.send_alert(settings, "Update Failed", f"Auto-update failed: {u_msg}") # Start Scheduler scheduler = BackgroundScheduler() def job_check_thresholds(): with app.app_context(): settings = Settings.query.first() if settings: monitor.check_thresholds_and_alert(settings) def job_run_backups(): with app.app_context(): jobs = BackupJob.query.all() for job in jobs: run_needed = False if not job.last_run: run_needed = True else: delta = datetime.now() - job.last_run if job.frequency == 'daily' and delta.days >= 1: run_needed = True elif job.frequency == 'weekly' and delta.days >= 7: run_needed = True if run_needed: success, msg = backup_manager.perform_backup_job(job) if success: job.last_run = datetime.now() db.session.commit() print(f"Backup job {job.name} succeeded") else: print(f"Backup job {job.name} failed: {msg}") if job.mysql_enabled: try: mysql_cfg = _get_mysql_backup_config(app.root_path) except Exception as cfg_err: mysql_cfg = {} print(f"MySQL backup config load error for job {job.name}: {cfg_err}") m_success, m_msg, m_path = backup_manager.perform_mysql_backup( mysql_cfg, db_mode=job.mysql_db_mode or 'all', db_name=job.mysql_db_name, db_list=job.mysql_db_list, backup_dir='/tmp/backups', prefix=f"mysql_job{job.id}", ) if m_success: print(f"MySQL backup for job {job.name} succeeded: {m_msg}") else: print(f"MySQL backup for job {job.name} failed: {m_msg}") if m_path and os.path.exists(m_path): os.remove(m_path) def job_update_traffic(): with app.app_context(): # Check if it's the first of the month to reset stats # (Simple logic: if last reset was previous month) # For now, simplistic: We just accumulate. # In real production, we'd have a 'last_reset' field. monitor.update_traffic_stats(db.session, Host) def job_check_quotas(): with app.app_context(): settings = Settings.query.first() if settings: monitor.check_quotas_and_lock(db.session, SystemUser, Host, settings) def job_check_updates(): with app.app_context(): settings = Settings.query.first() if settings: run_update_logic(settings) settings.last_update_check = datetime.now() db.session.commit() if not app.config.get('DISABLE_SCHEDULER'): scheduler.add_job(job_check_thresholds, 'interval', minutes=5, id='check_thresholds', name='System Health Check') scheduler.add_job(job_run_backups, 'interval', minutes=60, id='run_backups', name='Backup Scheduler') scheduler.add_job(job_update_traffic, 'interval', seconds=10, id='update_traffic', name='Traffic Stats Update') scheduler.add_job(job_check_quotas, 'interval', seconds=60, id='check_quotas', name='Quota Enforcement') scheduler.add_job(job_check_updates, 'interval', hours=24, id='check_updates', name='Auto Update Check') scheduler.start() scheduler_instance = scheduler @app.before_request def load_settings_and_check_ban(): app.config['WTF_CSRF_TRUSTED_ORIGINS'] = _trusted_origins_for_request() # Load settings globally for templates from flask import g try: g.settings = Settings.query.first() except: g.settings = None if not g.settings: # Fallback if DB is empty or failed g.settings = Settings(app_name='Limristem Web (Recovery)', logo_url='', favicon_url='', custom_message='') app.config['MAX_CONTENT_LENGTH'] = max( 1, _parse_non_negative_int(getattr(g.settings, 'panel_upload_limit_mb', 100), default=100, minimum=1, maximum=1024) ) * 1024 * 1024 # Inject Version g.version = VERSION # Load alerts for topbar and user if 'user_id' in session: try: g.user = Admin.query.get(session['user_id']) if g.user: g.unread_alerts = Alert.query.filter_by(is_read=False).count() g.recent_alerts = Alert.query.order_by(Alert.timestamp.desc()).limit(5).all() else: session.clear() g.unread_alerts = 0 g.recent_alerts = [] except Exception as e: print(f"Error loading user: {e}") session.clear() g.user = None g.unread_alerts = 0 g.recent_alerts = [] else: g.user = None g.unread_alerts = 0 g.recent_alerts = [] if request.endpoint == 'static': return ip = request.remote_addr ban = BannedIP.query.filter_by(ip_address=ip).first() if ban: if ban.banned_until > datetime.now(): return render_template('banned.html', ip=ip) else: db.session.delete(ban) db.session.commit() @app.context_processor def inject_global_stats(): lang = 'en' from flask import g if getattr(g, 'user', None) and g.user.language: lang = g.user.language elif request.accept_languages.best_match(['it', 'en', 'fr', 'es', 'de']) == 'it': lang = 'it' elif request.accept_languages.best_match(['it', 'en', 'fr', 'es', 'de']) == 'fr': lang = 'fr' elif request.accept_languages.best_match(['it', 'en', 'fr', 'es', 'de']) == 'es': lang = 'es' elif request.accept_languages.best_match(['it', 'en', 'fr', 'es', 'de']) == 'de': lang = 'de' try: stats = monitor.get_system_stats() with open('/proc/uptime', 'r') as f: uptime_seconds = float(f.readline().split()[0]) sys_uptime = f"{int(uptime_seconds // 86400)}d {int((uptime_seconds % 86400) // 3600)}h {int((uptime_seconds % 3600) // 60)}m" except Exception: stats = {} sys_uptime = "N/A" return dict(lang=lang, sys_stats=stats, sys_uptime=sys_uptime) @app.route('/') def index(): if 'user_id' not in session: return redirect(url_for('login')) return redirect(url_for('dashboard')) @app.route('/login', methods=('GET', 'POST')) def login(): if request.method == 'POST': username = request.form.get('username', '').strip() password = request.form.get('password', '') ip = request.remote_addr error = None user = Admin.query.filter_by(username=username).first() settings = Settings.query.first() if user is None or not check_password_hash(user.password, password): error = 'Incorrect username or password.' # Log Failure db.session.add(LoginAttempt(ip_address=ip, username=username, success=False)) db.session.commit() # Always create alert (will send email only if configured) monitor.send_alert(settings, "Security Alert: Failed Login", f"A failed login attempt for user '{username}' was detected from IP: {ip}", target_email=user.email if user and user.notify_on_failed_login else None) # Check for Ban recent_failures = LoginAttempt.query.filter( LoginAttempt.ip_address == ip, LoginAttempt.success == False, LoginAttempt.timestamp > datetime.now() - timedelta(minutes=5) ).count() # Calculate remaining attempts remaining = 5 - recent_failures if remaining <= 0: # Should be caught above, but for safety remaining = 0 if recent_failures >= 5: banned_until = datetime.now() + timedelta(minutes=30) db.session.add(BannedIP(ip_address=ip, banned_until=banned_until)) db.session.commit() if settings and settings.alert_email: monitor.send_alert(settings, "IP BANNED", f"IP {ip} has been banned for 30 minutes due to excessive failed logins.") return render_template('banned.html', ip=ip) # Render login with error and remaining attempts instead of flashing return render_template('login.html', error=error, remaining=remaining) # TOTP Check if user.totp_enabled: totp_code = request.form.get('totp_code', '') import pyotp if not totp_code or not pyotp.TOTP(user.totp_secret).verify(totp_code): # Show TOTP field return render_template('login.html', error="Invalid TOTP Code." if totp_code else None, require_totp=True, username=username, password=password) if error is None: session.clear() session['user_id'] = user.id # Log Success db.session.add(LoginAttempt(ip_address=ip, username=username, success=True)) db.session.commit() if settings and settings.alert_email: monitor.send_alert(settings, "Successful Login", f"User {username} logged in from IP: {ip}") if user.notify_on_login and user.email: monitor.send_alert(settings, "Security Alert: Successful Login", f"Your account {username} was accessed from IP: {ip}", target_email=user.email) # Check for Updates on Login if settings: has_update, new_ver, _ = update_manager.check_for_updates() if has_update: flash(f"Update Available: {new_ver}. Check Settings > Instance.", "info") session['available_update'] = new_ver msg = f"New version available: {new_ver}. Current: {VERSION}" existing_alert = Alert.query.filter(Alert.subject == "Update Available", Alert.message.like(f"%{new_ver}%")).first() if not existing_alert: monitor.send_alert(settings, "Update Available", msg) return redirect(url_for('dashboard')) # Should be handled by return above, but fallback return render_template('login.html', error=error) return render_template('login.html') @app.route('/logout') def logout(): session.clear() return redirect(url_for('login')) @app.route('/dashboard') def dashboard(): if 'user_id' not in session: return redirect(url_for('login')) stats = monitor.get_system_stats() recent_alerts = Alert.query.order_by(Alert.timestamp.desc()).limit(5).all() return render_template('dashboard.html', stats=stats, alerts=recent_alerts) @app.route('/alerts') def alerts_page(): if 'user_id' not in session: return redirect(url_for('login')) # Mark all as read try: Alert.query.filter_by(is_read=False).update({'is_read': True}) db.session.commit() except: db.session.rollback() all_alerts = Alert.query.order_by(Alert.timestamp.desc()).all() return render_template('alerts.html', alerts=all_alerts) @app.route('/php', methods=('GET', 'POST')) def php_page(): if 'user_id' not in session: return redirect(url_for('login')) if request.method == 'POST': # Handle Installation pass # TODO: call install logic versions = php_manager.get_installed_php_versions() return render_template('php.html', versions=versions) @app.route('/start_install_php', methods=['POST']) def start_install_php(): if 'user_id' not in session: return {'error': 'Unauthorized'}, 401 data = request.get_json(silent=True) or {} version = data.get('version') if not version: return {'error': 'Missing version'}, 400 cmd = php_manager.get_install_php_version_cmd(version) if not cmd: return {'error': 'Invalid version format'}, 400 task_id, err = task_manager.start_task(cmd) if not task_id: return {'error': err}, 500 return {'task_id': task_id} @app.route('/start_install_ext', methods=['POST']) def start_install_ext(): if 'user_id' not in session: return {'error': 'Unauthorized'}, 401 data = request.get_json(silent=True) or {} version = data.get('version') ext = data.get('extension') cmd = php_manager.get_install_php_extension_cmd(version, ext) if not cmd: return {'error': 'Invalid input'}, 400 task_id, err = task_manager.start_task(cmd) if not task_id: return {'error': err}, 500 return {'task_id': task_id} @app.route('/task_log/') def task_log(task_id): if 'user_id' not in session: return {'error': 'Unauthorized'}, 401 try: uuid.UUID(task_id) except ValueError: return {'error': 'Invalid task id'}, 400 content, finished, exit_code = task_manager.get_task_log(task_id) return { 'content': content, 'finished': finished, 'exit_code': exit_code } # Legacy route kept but redirects? Or just remove if we update UI fully. # We update UI to use AJAX. @app.route('/install_php', methods=['POST']) def install_php(): return redirect(url_for('php_page')) @app.route('/php//extensions') def php_extensions(version): if 'user_id' not in session: return redirect(url_for('login')) common = php_manager.get_common_extensions() return render_template('extensions.html', version=version, common_extensions=common) @app.route('/php//install_ext', methods=['POST']) def install_ext(version): # Fallback or redirect if JS disabled? # Ideally we use the AJAX route return redirect(url_for('php_extensions', version=version)) @app.route('/users', methods=('GET', 'POST')) def users_page(): if 'user_id' not in session: return redirect(url_for('login')) users = SystemUser.query.all() hosts = Host.query.all() return render_template('users.html', users=users, hosts=hosts) # Pass existing lists @app.route('/create_user', methods=['POST']) def create_user(): if 'user_id' not in session: return redirect(url_for('login')) username = request.form.get('username', '').strip() password = request.form.get('password', '') quota_mb = _parse_non_negative_int(request.form.get('quota_mb', 0)) email = request.form.get('email') cpu = _parse_non_negative_int(request.form.get('cpu_limit', 0)) ram = _parse_non_negative_int(request.form.get('ram_limit', 0)) io = _parse_non_negative_int(request.form.get('io_limit', 0)) if not is_valid_system_username(username): flash('Invalid username', 'error') return redirect(url_for('users_page')) if not password: flash('Password is required', 'error') return redirect(url_for('users_page')) success, msg = user_host_manager.create_system_user(username, password) if success: # Check if user exists in DB, if not add it existing_user = SystemUser.query.filter_by(username=username).first() if not existing_user: new_user = SystemUser( username=username, home_dir=f"/home/{username}", quota_limit_mb=quota_mb, email=email, cpu_limit_percent=cpu, ram_limit_mb=ram, io_limit_mb_s=io ) db.session.add(new_user) db.session.commit() existing_user = new_user # Scan for existing hosts discovered_hosts = user_host_manager.scan_existing_hosts(username) count = 0 for h in discovered_hosts: if not Host.query.filter_by(domain=h['domain']).first(): db.session.add(Host( domain=h['domain'], php_version=h['php_version'], root_dir=h['root_dir'], user_id=new_user.id )) count += 1 db.session.commit() flash(f'User imported. Discovered {count} sites.', 'success') else: flash('User updated/reset.', 'success') else: flash(f'Error: {msg}', 'error') return redirect(url_for('users_page')) @app.route('/edit_user/', methods=['POST']) def edit_user(user_id): if 'user_id' not in session: return redirect(url_for('login')) user = SystemUser.query.get(user_id) if not user: flash('User not found', 'error') return redirect(url_for('users_page')) password = request.form.get('password') email = request.form.get('email') quota_mb = _parse_non_negative_int(request.form.get('quota_mb', 0)) cpu = _parse_non_negative_int(request.form.get('cpu_limit', 0)) ram = _parse_non_negative_int(request.form.get('ram_limit', 0)) io = _parse_non_negative_int(request.form.get('io_limit', 0)) success, msg = user_host_manager.update_system_user(user.username, password, email, quota_mb, cpu, ram, io) if success: user.email = email user.quota_limit_mb = quota_mb user.cpu_limit_percent = cpu user.ram_limit_mb = ram user.io_limit_mb_s = io db.session.commit() flash('User updated successfully', 'success') else: flash(f'Update failed: {msg}', 'error') return redirect(url_for('users_page')) @app.route('/suspend_user/', methods=['POST']) def suspend_user(user_id): if 'user_id' not in session: return redirect(url_for('login')) user = SystemUser.query.get(user_id) if not user: return redirect(url_for('users_page')) action = request.form.get('action') if action == 'suspend': if not user.is_suspended: # System lock user_host_manager.suspend_system_user(user.username) # Suspend all hosts hosts = Host.query.filter_by(user_id=user.id).all() for h in hosts: user_host_manager.suspend_host(h.domain, 'account_suspended') h.is_suspended = True h.suspension_reason = 'account_suspended' user.is_suspended = True user.suspension_reason = 'manual' db.session.commit() flash(f'User {user.username} and all hosts suspended.', 'success') elif action == 'unsuspend': if user.is_suspended: # System unlock user_host_manager.unsuspend_system_user(user.username) # Unsuspend hosts (unless manually suspended before?) # For simplicity, we unsuspend all hosts suspended due to 'account_suspended' hosts = Host.query.filter_by(user_id=user.id).all() for h in hosts: if h.suspension_reason == 'account_suspended': user_host_manager.unsuspend_host( h.domain, h.php_version, user.home_dir, h.speed_limit_kbps, h.max_concurrent_connections, max_body_size_mb=h.nginx_max_body_size_mb, client_body_timeout=h.nginx_client_body_timeout, client_header_timeout=h.nginx_client_header_timeout, keepalive_timeout=h.nginx_keepalive_timeout ) h.is_suspended = False h.suspension_reason = None user.is_suspended = False user.suspension_reason = None db.session.commit() flash(f'User {user.username} unsuspended.', 'success') return redirect(url_for('users_page')) @app.route('/delete_user/', methods=['POST']) def delete_user(user_id): if 'user_id' not in session: return redirect(url_for('login')) user = SystemUser.query.get(user_id) if not user: return redirect(url_for('users_page')) # 1. Delete all hosts hosts = Host.query.filter_by(user_id=user.id).all() for h in hosts: user_host_manager.delete_host_system(h.domain, h.php_version, user.username) db.session.delete(h) # 2. Delete system user success, msg = user_host_manager.delete_system_user_complete(user.username, user.home_dir) if success: db.session.delete(user) db.session.commit() flash(f'User {user.username} deleted completely.', 'success') else: flash(f'Error deleting user: {msg}', 'error') return redirect(url_for('users_page')) @app.route('/hosts', methods=('GET', 'POST')) def hosts_page(): if 'user_id' not in session: return redirect(url_for('login')) users = SystemUser.query.all() hosts = Host.query.all() installed_php_versions = php_manager.get_installed_php_versions() # Sync SSL status from nginx config / cert files for each host ssl_changed = False for host in hosts: try: expiry_str, ssl_type = ssl_manager.get_domain_ssl_status(host.domain) if ssl_type and ssl_type not in ('No SSL', 'Config not found', 'Error checking'): if not host.ssl_enabled or host.ssl_provider != ssl_type: host.ssl_enabled = True host.ssl_provider = ssl_type ssl_changed = True parsed_expiry = None if expiry_str and expiry_str != "Active": try: # Format usually "notAfter=May 12 23:59:59 2024 GMT" -> we already split the notAfter part parsed_expiry = datetime.strptime(expiry_str.strip(), "%b %d %H:%M:%S %Y %Z") except Exception: pass if host.ssl_expiry != parsed_expiry: host.ssl_expiry = parsed_expiry ssl_changed = True elif ssl_type in ('No SSL', 'Config not found'): if host.ssl_enabled: host.ssl_enabled = False host.ssl_provider = None host.ssl_expiry = None ssl_changed = True except Exception: pass if ssl_changed: try: db.session.commit() except Exception: db.session.rollback() from models import CertificateAuthority cas = CertificateAuthority.query.all() return render_template('hosts.html', users=users, hosts=hosts, php_versions=installed_php_versions, cas=cas) @app.route('/create_host', methods=['POST']) def create_host(): if 'user_id' not in session: return redirect(url_for('login')) domain = (request.form.get('domain') or '').strip().lower().rstrip('.') user_id = request.form.get('user_id') php_version = request.form.get('php_version') host_type = request.form.get('host_type', 'php') template = request.form.get('template', 'default') if not domain or not user_id or not php_version: flash('Missing required fields.', 'error') return redirect(url_for('hosts_page')) if not is_valid_domain_name(domain): flash('Invalid domain name.', 'error') return redirect(url_for('hosts_page')) # Check DB first if Host.query.filter_by(domain=domain).first(): flash('Host already managed by Limristem Web.', 'error') return redirect(url_for('hosts_page')) # Check Nginx existence if user_host_manager.check_nginx_host_exists(domain): details = user_host_manager.get_nginx_host_details(domain) return render_template('import_host.html', domain=domain, detected_user=details.get('user'), php_version=details.get('php_version')) # Validate PHP version installed_versions = php_manager.get_installed_php_versions() if php_version not in installed_versions: flash(f'Error: PHP {php_version} is not installed.', 'error') return redirect(url_for('hosts_page')) # New Bandwidth Fields speed_limit = _parse_non_negative_int(request.form.get('speed_limit', 0)) traffic_limit = _parse_non_negative_int(request.form.get('traffic_limit', 0)) quota_mb = _parse_non_negative_int(request.form.get('quota_mb', 0)) conn_limit = _parse_non_negative_int(request.form.get('conn_limit', 10), default=10, minimum=1, maximum=10000) # Nginx timeout & payload settings (0 = default) ng_max_body = _parse_non_negative_int(request.form.get('ng_max_body', 0), default=0, minimum=0, maximum=1024) ng_body_timeout = _parse_non_negative_int(request.form.get('ng_body_timeout', 0), default=0, minimum=0, maximum=3600) ng_header_timeout = _parse_non_negative_int(request.form.get('ng_header_timeout', 0), default=0, minimum=0, maximum=3600) ng_keepalive = _parse_non_negative_int(request.form.get('ng_keepalive', 0), default=0, minimum=0, maximum=3600) user = SystemUser.query.get(user_id) if not user: flash('User not found', 'error') return redirect(url_for('hosts_page')) # Validate Quota Allocation if quota_mb > 0: if user.quota_limit_mb == 0: # User has unlimited, so host can have specific limit, allowed. pass else: # User has limit. Check available space. # existing_quota_sum = sum([h.quota_limit_mb for h in user.hosts]) # We need relation in model or simple query existing_hosts = Host.query.filter_by(user_id=user.id).all() used_quota = sum([h.quota_limit_mb for h in existing_hosts]) if (used_quota + quota_mb) > user.quota_limit_mb: flash(f'Error: Quota exceeds user limit. Available allocation: {user.quota_limit_mb - used_quota} MB', 'error') return redirect(url_for('hosts_page')) if host_type == 'php': success_pool, msg_pool = user_host_manager.create_php_pool(domain, php_version, user.username) if not success_pool: flash(f'Error creating PHP pool: {msg_pool}', 'error') return redirect(url_for('hosts_page')) success, msg = user_host_manager.create_nginx_host( domain, php_version, user.home_dir, speed_limit_kbps=speed_limit, conn_limit=conn_limit, host_type=host_type, template=template, max_body_size_mb=ng_max_body, client_body_timeout=ng_body_timeout, client_header_timeout=ng_header_timeout, keepalive_timeout=ng_keepalive ) if success: new_host = Host( domain=domain, php_version=php_version, root_dir=f"{user.home_dir}/www/{domain}/public", user_id=user.id, speed_limit_kbps=int(speed_limit), traffic_limit_mb=int(traffic_limit), quota_limit_mb=quota_mb, max_concurrent_connections=conn_limit, host_type=host_type, template=template, nginx_max_body_size_mb=ng_max_body, nginx_client_body_timeout=ng_body_timeout, nginx_client_header_timeout=ng_header_timeout, nginx_keepalive_timeout=ng_keepalive ) db.session.add(new_host) db.session.commit() flash('Host created with config', 'success') else: flash(f'Error: {msg}', 'error') return redirect(url_for('hosts_page')) @app.route('/import_host_confirm', methods=['POST']) def import_host_confirm(): if 'user_id' not in session: return redirect(url_for('login')) domain = (request.form.get('domain') or '').strip().lower().rstrip('.') detected_user = (request.form.get('detected_user') or '').strip() php_version = request.form['php_version'] if not is_valid_domain_name(domain): flash('Invalid domain name.', 'error') return redirect(url_for('hosts_page')) if not is_valid_system_username(detected_user): flash('Cannot import: invalid detected system user.', 'error') return redirect(url_for('hosts_page')) if not detected_user or detected_user == 'None': flash('Cannot import: User could not be detected from Nginx config.', 'error') return redirect(url_for('hosts_page')) # 1. Ensure User exists in Limristem Web user = SystemUser.query.filter_by(username=detected_user).first() if not user: # Create user entry # We assume user exists on system since Nginx config uses it new_user = SystemUser( username=detected_user, home_dir=f"/home/{detected_user}", quota_limit_mb=0 # Default unlimited for imported ) db.session.add(new_user) db.session.commit() user = new_user flash(f"System User '{detected_user}' imported into Limristem Web.", 'success') # 2. Create Host Entry if Host.query.filter_by(domain=domain).first(): flash('Host already managed.', 'warning') return redirect(url_for('hosts_page')) new_host = Host( domain=domain, php_version=php_version, root_dir=f"{user.home_dir}/www/{domain}/public", user_id=user.id, speed_limit_kbps=0, traffic_limit_mb=0, quota_limit_mb=0, max_concurrent_connections=10 ) db.session.add(new_host) db.session.commit() flash(f"Host '{domain}' imported successfully.", 'success') return redirect(url_for('hosts_page')) @app.route('/get_host_config/') def get_host_config(host_id): if 'user_id' not in session: return jsonify({'error': 'Unauthorized'}), 401 try: host = Host.query.get(host_id) if not host: return jsonify({'error': 'Not found'}), 404 content = user_host_manager.get_nginx_config_content(host.domain) return jsonify({'content': content}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/save_host_config/', methods=['POST']) def save_host_config(host_id): if 'user_id' not in session: return {'error': 'Unauthorized'}, 401 host = Host.query.get(host_id) if not host: return {'error': 'Not found'}, 404 data = request.get_json(silent=True) or {} content = data.get('content') if not content: return {'error': 'Empty content'}, 400 success, msg = user_host_manager.save_nginx_config_content(host.domain, content) if success: return {'status': 'saved'} else: return {'error': msg}, 500 @app.route('/update_host_settings/', methods=['POST']) def update_host_settings(host_id): if 'user_id' not in session: return redirect(url_for('login')) host = Host.query.get(host_id) if not host: flash('Host not found', 'error') return redirect(url_for('hosts_page')) new_php = request.form.get('php_version') speed = _parse_non_negative_int(request.form.get('speed_limit', 0)) traffic = _parse_non_negative_int(request.form.get('traffic_limit', 0)) quota_mb = _parse_non_negative_int(request.form.get('quota_mb', 0)) conn_limit = _parse_non_negative_int(request.form.get('conn_limit', 10), default=10, minimum=1, maximum=10000) host_type = request.form.get('host_type', host.host_type) template = request.form.get('template', host.template) # Nginx timeout & payload settings (0 = default) ng_max_body = _parse_non_negative_int(request.form.get('ng_max_body', 0), default=0, minimum=0, maximum=1024) ng_body_timeout = _parse_non_negative_int(request.form.get('ng_body_timeout', 0), default=0, minimum=0, maximum=3600) ng_header_timeout = _parse_non_negative_int(request.form.get('ng_header_timeout', 0), default=0, minimum=0, maximum=3600) ng_keepalive = _parse_non_negative_int(request.form.get('ng_keepalive', 0), default=0, minimum=0, maximum=3600) # Immediate sync triggering force_check = False if quota_mb > host.quota_limit_mb or traffic > host.traffic_limit_mb: force_check = True # Validate PHP version installed_versions = php_manager.get_installed_php_versions() if new_php not in installed_versions: flash(f'Error: PHP {new_php} is not installed.', 'error') return redirect(url_for('hosts_page')) user = SystemUser.query.get(host.user_id) # Validate Quota if quota_mb > 0: if user.quota_limit_mb > 0: existing_hosts = Host.query.filter(Host.user_id == user.id, Host.id != host.id).all() used_quota = sum([h.quota_limit_mb for h in existing_hosts]) if (used_quota + quota_mb) > user.quota_limit_mb: flash(f'Error: Quota exceeds user limit.', 'error') return redirect(url_for('hosts_page')) # Handle PHP Pool Changes if host_type == 'php': # Validate php_pm against whitelist to prevent FPM config injection php_pm = request.form.get('php_pm', 'ondemand') if not is_valid_php_pm(php_pm): php_pm = 'ondemand' # Validate memory_limit and upload_max (PHP shorthand bytes format) php_memory_limit = request.form.get('php_memory_limit', '128M') if not is_valid_php_size(php_memory_limit): php_memory_limit = '128M' php_upload_max = request.form.get('php_upload_max_filesize', '10M') if not is_valid_php_size(php_upload_max): php_upload_max = '10M' # post_max_size should be >= upload_max_filesize php_post_max = php_upload_max php_settings = { 'pm': php_pm, 'max_children': _parse_non_negative_int(request.form.get('php_max_children', 5), default=5, minimum=1, maximum=1024), 'start_servers': 2, 'min_spare': 1, 'max_spare': 3, 'memory_limit': php_memory_limit, 'upload_max': php_upload_max, 'post_max': php_post_max, 'max_exec': _parse_non_negative_int(request.form.get('php_max_execution_time', 30), default=30, minimum=1, maximum=3600) } # Ensure pool is updated/created user_host_manager.create_php_pool(host.domain, new_php, user.username, php_settings) else: # If switching to HTML, remove pool if host.host_type == 'php': user_host_manager.remove_php_pool(host.domain, host.php_version) php_settings = None # Re-generate Nginx Config success, msg = user_host_manager.create_nginx_host( host.domain, new_php, user.home_dir, speed_limit_kbps=speed, conn_limit=conn_limit, host_type=host_type, template=template, max_body_size_mb=ng_max_body, client_body_timeout=ng_body_timeout, client_header_timeout=ng_header_timeout, keepalive_timeout=ng_keepalive ) if success: host.php_version = new_php host.speed_limit_kbps = speed host.traffic_limit_mb = traffic host.quota_limit_mb = quota_mb host.max_concurrent_connections = conn_limit host.host_type = host_type host.template = template host.nginx_max_body_size_mb = ng_max_body host.nginx_client_body_timeout = ng_body_timeout host.nginx_client_header_timeout = ng_header_timeout host.nginx_keepalive_timeout = ng_keepalive if php_settings: host.php_pm = php_settings['pm'] host.php_max_children = php_settings['max_children'] host.php_memory_limit = php_settings['memory_limit'] host.php_upload_max_filesize = php_settings['upload_max'] host.php_max_execution_time = php_settings['max_exec'] db.session.commit() if force_check: settings = Settings.query.first() monitor.check_quotas_and_lock(db.session, SystemUser, Host, settings) flash('Host settings updated successfully', 'success') else: flash(f'Update failed: {msg}', 'error') return redirect(url_for('hosts_page')) @app.route('/toggle_host_status/', methods=['POST']) def toggle_host_status(host_id): if 'user_id' not in session: return redirect(url_for('login')) host = Host.query.get(host_id) if not host: return redirect(url_for('hosts_page')) action = request.form.get('action') user = SystemUser.query.get(host.user_id) if action == 'suspend': if not host.is_suspended: user_host_manager.suspend_host(host.domain, 'suspended') host.is_suspended = True host.suspension_reason = 'manual' db.session.commit() flash(f'Host {host.domain} suspended.', 'success') elif action == 'unsuspend': if host.is_suspended: 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() flash(f'Host {host.domain} unsuspended.', 'success') return redirect(url_for('hosts_page')) @app.route('/refresh_host/', methods=['POST']) def refresh_host(host_id): if 'user_id' not in session: return redirect(url_for('login')) host = Host.query.get(host_id) if not host: flash('Host not found', 'error') return redirect(url_for('hosts_page')) user = SystemUser.query.get(host.user_id) 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() settings = Settings.query.first() monitor.check_quotas_and_lock(db.session, SystemUser, Host, settings) monitor.update_traffic_stats(db.session, Host) flash('Host configuration refreshed and limits re-checked.', 'success') return redirect(url_for('hosts_page')) @app.route('/delete_host/', methods=['POST']) def delete_host(host_id): if 'user_id' not in session: return redirect(url_for('login')) host = Host.query.get(host_id) if not host: flash('Host not found', 'error') return redirect(url_for('hosts_page')) user = SystemUser.query.get(host.user_id) # Delete system resources success, msg = user_host_manager.delete_host_system(host.domain, host.php_version, user.username) if success: db.session.delete(host) db.session.commit() flash('Host deleted successfully', 'success') else: flash(f'Error deleting host system resources: {msg}', 'error') return redirect(url_for('hosts_page')) @app.route('/update_host_ssl/', methods=['POST']) def update_host_ssl(host_id): if 'user_id' not in session: return redirect(url_for('login')) host = Host.query.get(host_id) if not host: flash('Host not found', 'error') return redirect(url_for('hosts_page')) ssl_type = request.form.get('ssl_type') settings = Settings.query.first() admin_email = settings.alert_email if (settings and settings.alert_email) else 'admin@example.com' if ssl_type in ['self', 'letsencrypt']: script_path = os.path.join(app.root_path, 'run_ssl_task.py') import sys arg3 = admin_email if ssl_type == 'self': ca_id_val = request.form.get('ca_id') if ca_id_val == 'new_ca': import secrets from models import CertificateAuthority ca_name = f"CA_{host.domain.replace('.', '_')}_{secrets.token_hex(4)}" success, crt, key = ssl_manager.generate_ca(ca_name, host.domain) if success: new_ca = CertificateAuthority(name=ca_name, cert_path=crt, key_path=key) db.session.add(new_ca) db.session.commit() arg3 = str(new_ca.id) elif ca_id_val: arg3 = ca_id_val else: arg3 = "" cmd = [sys.executable, script_path, str(host.id), ssl_type, arg3] task_id, err = task_manager.start_task(cmd) if not task_id: if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return {'error': err}, 500 flash(f'Failed to start SSL task: {err}', 'error') return redirect(url_for('hosts_page')) if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return {'task_id': task_id} flash('SSL configuration started in background.', 'info') return redirect(url_for('hosts_page')) else: if ssl_type == 'manual': cert = request.form.get('cert_content') key = request.form.get('key_content') if cert and key: success, msg = ssl_manager.enable_ssl_manual(host.domain, cert, key) else: success, msg = False, "Certificate and Key required" else: success, msg = False, "Disable not implemented yet" if success: host.ssl_enabled = True host.ssl_provider = ssl_type db.session.commit() flash(f'SSL Updated: {msg}', 'success') else: flash(f'SSL Error: {msg}', 'error') return redirect(url_for('hosts_page')) @app.route('/backups', methods=('GET', 'POST')) def backups_page(): if 'user_id' not in session: return redirect(url_for('login')) jobs = BackupJob.query.all() return render_template('backups.html', jobs=jobs) @app.route('/cron') def cron_page(): if 'user_id' not in session: return redirect(url_for('login')) jobs = [] if scheduler_instance: for job in scheduler_instance.get_jobs(): # Extract interval value and unit from trigger for display/edit interval_value = 5 interval_unit = 'minutes' try: t = job.trigger # APScheduler IntervalTrigger stores interval as a timedelta if hasattr(t, 'interval'): td = t.interval total_seconds = int(td.total_seconds()) if total_seconds % 604800 == 0: interval_value = total_seconds // 604800 interval_unit = 'weeks' elif total_seconds % 86400 == 0: interval_value = total_seconds // 86400 interval_unit = 'days' elif total_seconds % 3600 == 0: interval_value = total_seconds // 3600 interval_unit = 'hours' elif total_seconds % 60 == 0: interval_value = total_seconds // 60 interval_unit = 'minutes' else: interval_value = total_seconds interval_unit = 'seconds' except Exception: pass jobs.append({ 'id': job.id, 'name': job.name, 'next_run_time': job.next_run_time, 'trigger': str(job.trigger), 'interval_value': interval_value, 'interval_unit': interval_unit, }) return render_template('cron.html', jobs=jobs) @app.route('/trigger_cron_job', methods=['POST']) def trigger_cron_job(): if 'user_id' not in session: return redirect(url_for('login')) job_id = request.form.get('job_id') if scheduler_instance and job_id: try: job = scheduler_instance.get_job(job_id) if job: job.modify(next_run_time=datetime.now()) flash(f"Job '{job.name}' triggered successfully.", "success") else: flash(f"Job not found.", "error") except Exception as e: flash(f"Error triggering job: {e}", "error") return redirect(url_for('cron_page')) @app.route('/edit_cron_job', methods=['POST']) def edit_cron_job(): if 'user_id' not in session: return redirect(url_for('login')) job_id = request.form.get('job_id') new_name = request.form.get('job_name', '').strip() try: interval_value = int(request.form.get('interval_value', 5)) except (ValueError, TypeError): interval_value = 5 interval_unit = request.form.get('interval_unit', 'minutes') if interval_unit not in ('seconds', 'minutes', 'hours', 'days', 'weeks'): interval_unit = 'minutes' if scheduler_instance and job_id: try: job = scheduler_instance.get_job(job_id) if job: if new_name: job.modify(name=new_name) kwargs = {interval_unit: interval_value} job.reschedule(trigger='interval', **kwargs) flash(f"Job '{new_name or job.name}' updated successfully.", "success") else: flash("Job not found.", "error") except Exception as e: flash(f"Error updating job: {e}", "error") return redirect(url_for('cron_page')) @app.route('/logs') def logs_page(): if 'user_id' not in session: return redirect(url_for('login')) # System/console logs from instance logs dir log_dir = os.path.join(app.instance_path, 'logs') system_logs = [] if os.path.exists(log_dir): system_logs = sorted( [f for f in os.listdir(log_dir) if os.path.isfile(os.path.join(log_dir, f))], reverse=True ) # Nginx logs per host domain (always list expected paths; read endpoint handles missing files) hosts = Host.query.all() nginx_logs = {} for host in hosts: files = [ {'label': 'access.log', 'path': f"/var/log/nginx/{host.domain}.access.log"}, {'label': 'error.log', 'path': f"/var/log/nginx/{host.domain}.error.log"}, ] nginx_logs[host.domain] = files # PHP-FPM logs: global per-version log + per-host error log php_logs = {} installed_php = php_manager.get_installed_php_versions() for ver in installed_php: global_path = f"/var/log/php{ver}-fpm.log" host_logs = [ { 'domain': h.domain, 'path': f"/var/log/php{ver}-fpm/{h.domain}.error.log" } for h in hosts if h.php_version == ver ] php_logs[ver] = { 'global': global_path, 'hosts': host_logs, } return render_template('logs.html', system_logs=system_logs, nginx_logs=nginx_logs, php_logs=php_logs, ) @app.route('/read_log_file') def read_log_file(): if 'user_id' not in session: return jsonify({'error': 'Unauthorized'}), 401 filename = request.args.get('filename') if not filename: return jsonify({'error': 'Missing filename'}), 400 if filename == 'system:limristem-web': try: import subprocess result = subprocess.run(['journalctl', '-u', 'limristem-web', '-n', '500', '--no-pager'], capture_output=True, text=True) return jsonify({'content': result.stdout}) except Exception as e: return jsonify({'error': f"Failed to fetch journal: {e}"}), 500 filepath = _is_allowed_log_path(app, filename) if not filepath: return jsonify({'error': 'Access denied'}), 403 if not os.path.exists(filepath): return jsonify({'error': 'File not found'}), 404 try: with open(filepath, 'rb') as f: # Read last 50KB to avoid massive payloads f.seek(0, 2) size = f.tell() f.seek(max(size - 50000, 0)) content = f.read().decode('utf-8', errors='replace') return jsonify({'content': content}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/create_backup', methods=['POST']) def create_backup(): if 'user_id' not in session: return redirect(url_for('login')) name = request.form['name'] source = request.form['source_path'] dtype = request.form['dest_type'] freq = request.form['frequency'] ret_days = _parse_non_negative_int(request.form.get('retention_days', 30), default=30) ret_count = _parse_non_negative_int(request.form.get('retention_count', 10), default=10) if not os.path.isabs(source) or not os.path.exists(source): flash('Source path must be an existing absolute path.', 'error') return redirect(url_for('backups_page')) if dtype not in {'local', 'sftp', 's3'}: flash('Invalid backup destination type.', 'error') return redirect(url_for('backups_page')) if freq not in {'daily', 'weekly'}: flash('Invalid backup frequency.', 'error') return redirect(url_for('backups_page')) # Build Config config = {} if dtype == 'local': config['path'] = request.form.get('local_path') if not config['path'] or not os.path.isabs(config['path']): flash('Local backup destination must be an absolute path.', 'error') return redirect(url_for('backups_page')) elif dtype == 'sftp': config['host'] = request.form.get('sftp_host') config['port'] = _parse_non_negative_int(request.form.get('sftp_port', 22), default=22, minimum=1, maximum=65535) config['user'] = request.form.get('sftp_user') config['pass'] = request.form.get('sftp_pass') config['key'] = request.form.get('sftp_key') config['remote_path'] = request.form.get('sftp_path') elif dtype == 's3': config['endpoint'] = request.form.get('s3_endpoint') config['bucket'] = request.form.get('s3_bucket') config['region'] = request.form.get('s3_region') config['access_key'] = request.form.get('s3_access_key') config['secret_key'] = request.form.get('s3_secret_key') config_json = json.dumps(config) # MySQL Backup settings mysql_enabled = request.form.get('mysql_enabled') == '1' mysql_db_mode = request.form.get('mysql_db_mode', 'all') mysql_db_name = request.form.get('mysql_db_name', '').strip() mysql_db_list = request.form.get('mysql_db_list', '').strip() mysql_err = _validate_mysql_backup_settings(mysql_enabled, mysql_db_mode, mysql_db_name, mysql_db_list) if mysql_err: flash(mysql_err, 'error') return redirect(url_for('backups_page')) job = BackupJob( name=name, source_path=source, dest_type=dtype, dest_config=config_json, frequency=freq, retention_days=ret_days, retention_count=ret_count, mysql_enabled=mysql_enabled, mysql_db_mode=mysql_db_mode, mysql_db_name=mysql_db_name or None, mysql_db_list=mysql_db_list or None, ) db.session.add(job) db.session.commit() flash('Backup job saved', 'success') return redirect(url_for('backups_page')) @app.route('/run_backup/', methods=['POST']) def run_backup(job_id): if 'user_id' not in session: return redirect(url_for('login')) job = BackupJob.query.get(job_id) if job: success, msg = backup_manager.perform_backup_job(job) mysql_success = True if job.mysql_enabled: mysql_cfg = _get_mysql_backup_config(app.root_path) m_success, m_msg, m_path = backup_manager.perform_mysql_backup( mysql_cfg, db_mode=job.mysql_db_mode or 'all', db_name=job.mysql_db_name, db_list=job.mysql_db_list, backup_dir='/tmp/backups', prefix=f"mysql_job{job.id}", ) mysql_success = m_success if m_success: print(f"MySQL backup for job '{job.name}' succeeded: {m_msg}") else: print(f"MySQL backup for job '{job.name}' failed: {m_msg}") if m_path and os.path.exists(m_path): os.remove(m_path) if success and mysql_success: job.last_run = datetime.now() db.session.commit() flash(f"Backup job '{job.name}' executed successfully.", "success") elif success and not mysql_success: job.last_run = datetime.now() db.session.commit() flash(f"Backup job '{job.name}' files OK but MySQL backup failed: {m_msg}", "warning") else: flash(f"Backup job '{job.name}' failed: {msg}", "error") else: flash("Job not found", "error") return redirect(url_for('backups_page')) @app.route('/edit_backup/', methods=['POST']) def edit_backup(job_id): if 'user_id' not in session: return redirect(url_for('login')) job = BackupJob.query.get(job_id) if not job: flash('Backup job not found', 'error') return redirect(url_for('backups_page')) job.name = request.form.get('name', job.name) job.source_path = request.form.get('source_path', job.source_path) job.frequency = request.form.get('frequency', job.frequency) if not os.path.isabs(job.source_path) or not os.path.exists(job.source_path): flash('Source path must be an existing absolute path.', 'error') return redirect(url_for('backups_page')) try: job.retention_days = int(request.form.get('retention_days', job.retention_days)) job.retention_count = int(request.form.get('retention_count', job.retention_count)) except (ValueError, TypeError): pass dtype = request.form.get('dest_type', job.dest_type) job.dest_type = dtype config = {} try: existing = json.loads(job.dest_config or '{}') except (json.JSONDecodeError, TypeError): existing = {} if dtype == 'local': config['path'] = request.form.get('local_path', '') if not config['path'] or not os.path.isabs(config['path']): flash('Local backup destination must be an absolute path.', 'error') return redirect(url_for('backups_page')) elif dtype == 'sftp': config['host'] = request.form.get('sftp_host', '') config['port'] = _parse_non_negative_int(request.form.get('sftp_port', '22'), default=22, minimum=1, maximum=65535) config['user'] = request.form.get('sftp_user', '') sftp_pass = request.form.get('sftp_pass', '') config['pass'] = sftp_pass if sftp_pass else existing.get('pass', '') config['key'] = request.form.get('sftp_key', '') config['remote_path'] = request.form.get('sftp_path', '') elif dtype == 's3': config['endpoint'] = request.form.get('s3_endpoint', '') config['bucket'] = request.form.get('s3_bucket', '') config['region'] = request.form.get('s3_region', '') config['access_key'] = request.form.get('s3_access_key', '') s3_secret = request.form.get('s3_secret_key', '') config['secret_key'] = s3_secret if s3_secret else existing.get('secret_key', '') job.dest_config = json.dumps(config) # MySQL Backup settings job.mysql_enabled = request.form.get('mysql_enabled') == '1' mysql_db_mode = request.form.get('mysql_db_mode', 'all') mysql_db_name = request.form.get('mysql_db_name', '').strip() mysql_db_list = request.form.get('mysql_db_list', '').strip() mysql_err = _validate_mysql_backup_settings(job.mysql_enabled, mysql_db_mode, mysql_db_name, mysql_db_list) if mysql_err: flash(mysql_err, 'error') return redirect(url_for('backups_page')) job.mysql_db_mode = mysql_db_mode job.mysql_db_name = mysql_db_name or None job.mysql_db_list = mysql_db_list or None db.session.commit() flash('Backup job updated successfully', 'success') return redirect(url_for('backups_page')) @app.route('/databases') def databases_page(): if 'user_id' not in session: return redirect(url_for('login')) dbs, err1 = mysql_manager.list_databases() users, err2 = mysql_manager.list_users() setup_needed = False if err1 and ("config not found" in err1 or "Access denied" in err1 or "Can't connect" in err1): setup_needed = True return render_template('databases.html', databases=dbs, db_users=users, setup_needed=setup_needed, setup_error=err1) @app.route('/start_mysql_install', methods=['POST']) def start_mysql_install(): if 'user_id' not in session: return {'error': 'Unauthorized'}, 401 # Determine path to helper script script_path = os.path.join(app.root_path, 'install_mysql_task.py') cmd = [sys.executable, script_path] task_id, err = task_manager.start_task(cmd) if not task_id: return {'error': err}, 500 return {'task_id': task_id} # Legacy route removed/redirected @app.route('/setup_mysql_install', methods=['POST']) def setup_mysql_install(): return redirect(url_for('databases_page')) @app.route('/setup_mysql_config', methods=['POST']) def setup_mysql_config(): if 'user_id' not in session: return redirect(url_for('login')) password = request.form.get('root_password') if password: mysql_manager.configure_mysql_connection(password) # Validate _, err = mysql_manager.list_databases() if not err: flash("MySQL Configured Successfully", 'success') else: flash(f"Configuration saved but connection failed: {err}", 'warning') return redirect(url_for('databases_page')) @app.route('/create_db', methods=['POST']) def create_db(): if 'user_id' not in session: return redirect(url_for('login')) dbname = request.form.get('dbname') success, msg = mysql_manager.create_database(dbname) if success: flash(msg, 'success') else: flash(msg, 'error') return redirect(url_for('databases_page')) @app.route('/create_db_user', methods=['POST']) def create_db_user(): if 'user_id' not in session: return redirect(url_for('login')) username = request.form.get('username') password = request.form.get('password') success, msg = mysql_manager.create_user(username, password) if success: flash(msg, 'success') else: flash(msg, 'error') return redirect(url_for('databases_page')) @app.route('/grant_db_privileges', methods=['POST']) def grant_db_privileges(): if 'user_id' not in session: return redirect(url_for('login')) dbname = request.form.get('dbname') username = request.form.get('username') success, msg = mysql_manager.grant_privileges(dbname, username) if success: flash(msg, 'success') else: flash(msg, 'error') return redirect(url_for('databases_page')) @app.route('/settings', methods=('GET', 'POST')) def settings_page(): if 'user_id' not in session: return redirect(url_for('login')) if 'available_update' in session: if not update_manager.is_newer(session['available_update'], VERSION): session.pop('available_update', None) settings = Settings.query.first() from models import CertificateAuthority cas = CertificateAuthority.query.all() return render_template('settings.html', settings=settings, cas=cas) @app.route('/profile') def profile_page(): if 'user_id' not in session: return redirect(url_for('login')) try: import pyotp totp_secret = getattr(g.user, 'totp_secret', None) if not totp_secret: totp_secret = pyotp.random_base32() totp_uri = pyotp.totp.TOTP(totp_secret).provisioning_uri(name=g.user.username, issuer_name="Limristem Web") return render_template('profile.html', user=g.user, totp_secret=totp_secret, totp_uri=totp_uri) except ImportError: flash("PyOTP module is not installed. Please run pip install pyotp", "error") return render_template('profile.html', user=g.user, totp_secret=None, totp_uri=None) @app.route('/update_profile', methods=['POST']) def update_profile(): if 'user_id' not in session: return redirect(url_for('login')) user = Admin.query.get(session['user_id']) form_action = request.form.get('form_action') if form_action == 'general': user.nickname = request.form.get('nickname') user.email = request.form.get('email') user.timezone = request.form.get('timezone', 'UTC') user.language = request.form.get('language', 'en') user.notify_on_login = 'notify_on_login' in request.form user.notify_on_failed_login = 'notify_on_failed_login' in request.form user.notify_update_available = 'notify_update_available' in request.form user.notify_update_installed = 'notify_update_installed' in request.form flash('General profile updated successfully', 'success') elif form_action == 'password': current_password = request.form.get('current_password') new_password = request.form.get('new_password') confirm_password = request.form.get('confirm_password') if not check_password_hash(user.password, current_password): flash('Current password is incorrect', 'error') elif new_password != confirm_password: flash('New passwords do not match', 'error') elif not new_password: flash('New password cannot be empty', 'error') else: user.password = generate_password_hash(new_password) flash('Password updated successfully', 'success') elif form_action == 'avatar': if 'avatar_file' in request.files and not request.form.get('remove_avatar'): file = request.files['avatar_file'] if file and file.filename != '': try: # 1MB limit = 1024KB filename = _save_uploaded_asset(file, app.config['UPLOAD_FOLDER'], f'avatar_{user.id}', max_size_kb=1024) user.avatar_url = url_for('uploaded_file', filename=filename) flash('Avatar updated successfully', 'success') except ValueError as e: flash(str(e), 'error') elif request.form.get('remove_avatar'): user.avatar_url = None flash('Avatar removed successfully', 'success') # TOTP Settings elif request.form.get('totp_action'): action = request.form.get('totp_action') if action == 'enable': secret = request.form.get('totp_secret') code = request.form.get('totp_code') try: import pyotp if pyotp.TOTP(secret).verify(code): user.totp_secret = secret user.totp_enabled = True flash('TOTP enabled successfully', 'success') else: flash('Invalid TOTP code', 'error') except ImportError: flash('PyOTP module not installed', 'error') elif action == 'disable': user.totp_enabled = False user.totp_secret = None flash('TOTP disabled', 'success') db.session.commit() return redirect(url_for('profile_page')) @app.route('/uploads/') def uploaded_file(filename): safe_name = secure_filename(filename) if not safe_name: return jsonify({'error': 'Invalid filename'}), 400 return send_from_directory(app.config['UPLOAD_FOLDER'], safe_name) @app.route('/update_settings', methods=['POST']) def update_settings(): if 'user_id' not in session: return redirect(url_for('login')) settings = Settings.query.first() # General settings.app_name = (request.form.get('app_name', 'Limristem Web') or 'Limristem Web').strip()[:100] logo_url = (request.form.get('logo_url') or '').strip() favicon_url = (request.form.get('favicon_url') or '').strip() if logo_url and not is_safe_external_url(logo_url, allow_relative=True): flash('Invalid logo URL.', 'error') return redirect(url_for('settings_page')) if favicon_url and not is_safe_external_url(favicon_url, allow_relative=True): flash('Invalid favicon URL.', 'error') return redirect(url_for('settings_page')) if request.form.get('remove_logo'): settings.logo_url = None else: settings.logo_url = logo_url or None if request.form.get('remove_favicon'): settings.favicon_url = None else: settings.favicon_url = favicon_url or None settings.custom_message = request.form.get('custom_message') # New Hostname Handling new_hostname = (request.form.get('server_hostname') or '').strip() if new_hostname and not is_valid_hostname(new_hostname): flash('Invalid system hostname.', 'error') return redirect(url_for('settings_page')) if new_hostname and new_hostname != settings.server_hostname: success, msg = user_host_manager.update_system_hostname(new_hostname) if success: # Regenerate Panel Config user_host_manager.regenerate_panel_nginx_config( new_hostname, ssl_type=settings.panel_ssl_type or 'self', install_dir=os.path.dirname(app.root_path) # Assuming root_path is limristem_web/ -> ../ ) settings.server_hostname = new_hostname flash(f"Hostname updated to {new_hostname}. Nginx reloaded.", "success") else: flash(f"Failed to update hostname: {msg}", "error") # File Uploads if 'logo_file' in request.files and not request.form.get('remove_logo'): file = request.files['logo_file'] if file and file.filename != '': try: filename = _save_uploaded_asset(file, app.config['UPLOAD_FOLDER'], 'logo', max_size_kb=512) settings.logo_url = url_for('uploaded_file', filename=filename) except ValueError as e: flash(str(e), 'error') return redirect(url_for('settings_page')) if 'favicon_file' in request.files and not request.form.get('remove_favicon'): file = request.files['favicon_file'] if file and file.filename != '': try: filename = _save_uploaded_asset(file, app.config['UPLOAD_FOLDER'], 'favicon', max_size_kb=32) settings.favicon_url = url_for('uploaded_file', filename=filename) except ValueError as e: flash(str(e), 'error') return redirect(url_for('settings_page')) # Control settings.smtp_server = request.form.get('smtp_server') settings.smtp_port = _parse_non_negative_int(request.form.get('smtp_port'), default=0, minimum=0, maximum=65535) or None settings.smtp_user = request.form.get('smtp_user') smtp_pass = request.form.get('smtp_pass') if smtp_pass: settings.smtp_pass = smtp_pass settings.alert_email = request.form.get('alert_email') webhook_url = (request.form.get('webhook_url') or '').strip() if webhook_url and not is_safe_external_url(webhook_url, block_internal=True): flash('Invalid webhook URL.', 'error') return redirect(url_for('settings_page')) settings.webhook_url = webhook_url or None webhook_headers = (request.form.get('webhook_headers') or '').strip() if webhook_headers: try: parsed_headers = json.loads(webhook_headers) if not isinstance(parsed_headers, dict): raise ValueError() except ValueError: flash('Webhook headers must be a valid JSON object.', 'error') return redirect(url_for('settings_page')) settings.webhook_headers = webhook_headers or None # Panel SSL Logic panel_ssl_type = request.form.get('panel_ssl_type') if panel_ssl_type and panel_ssl_type != settings.panel_ssl_type: settings.panel_ssl_type = panel_ssl_type # Apply Change if panel_ssl_type == 'letsencrypt': if settings.server_hostname and settings.alert_email: # Async task preferred but sync for MVP # This might block! success, msg = ssl_manager.enable_panel_letsencrypt(settings.server_hostname, settings.alert_email) if success: flash("Let's Encrypt configured for Panel.", "success") else: flash(f"Let's Encrypt Error: {msg}", "error") else: flash("Let's Encrypt requires Hostname and Alert Email to be set first.", "warning") elif panel_ssl_type == 'self': # Regenerate Self Signed if needed ca_id_val = request.form.get('ca_id') ca_cert_path = None ca_key_path = None if ca_id_val == 'new_ca': import secrets from models import CertificateAuthority ca_name = f"CA_{settings.server_hostname.replace('.', '_')}_{secrets.token_hex(4)}" c_succ, c_crt, c_key = ssl_manager.generate_ca(ca_name, settings.server_hostname) if c_succ: new_ca = CertificateAuthority(name=ca_name, cert_path=c_crt, key_path=c_key) db.session.add(new_ca) db.session.commit() ca_cert_path = c_crt ca_key_path = c_key elif ca_id_val and ca_id_val.isdigit(): from models import CertificateAuthority ca = CertificateAuthority.query.get(int(ca_id_val)) if ca: ca_cert_path = ca.cert_path ca_key_path = ca.key_path succ, msg = ssl_manager.generate_panel_selfsigned( settings.server_hostname or 'localhost', os.path.dirname(app.root_path), ca_cert_path, ca_key_path ) if succ: flash('Self-Signed Certificate regenerated for Panel.', 'success') user_host_manager.regenerate_panel_nginx_config( settings.server_hostname or 'localhost', ssl_type='self', install_dir=os.path.dirname(app.root_path) ) else: flash(f'Self-Signed Error: {msg}', 'error') # Panel Cert Uploads (Manual) if panel_ssl_type == 'manual': if 'panel_cert' in request.files and 'panel_key' in request.files: cert_file = request.files['panel_cert'] key_file = request.files['panel_key'] if cert_file and key_file and cert_file.filename != '' and key_file.filename != '': try: cert_path = os.path.join(app.instance_path, 'panel.crt') key_path = os.path.join(app.instance_path, 'panel.key') cert_file.save(cert_path) key_file.save(key_path) flash('Panel SSL Certificate uploaded.', 'success') # Trigger reload user_host_manager.regenerate_panel_nginx_config( settings.server_hostname or 'localhost', ssl_type='manual', install_dir=os.path.dirname(app.root_path) ) except Exception as e: flash(f'Failed to update Panel SSL: {e}', 'error') # Panel Limits & Config panel_limit = _parse_non_negative_int(request.form.get('panel_upload_limit_mb', 100), default=100, minimum=1, maximum=1024) panel_timeout = _parse_non_negative_int(request.form.get('panel_timeout_seconds', 300), default=300, minimum=10, maximum=3600) if panel_limit != settings.panel_upload_limit_mb or panel_timeout != settings.panel_timeout_seconds: settings.panel_upload_limit_mb = panel_limit settings.panel_timeout_seconds = panel_timeout # Update System Config success, msg = user_host_manager.update_panel_config(panel_limit, panel_timeout) if not success: flash(f"Warning: Panel config saved but system update failed: {msg}", "warning") else: flash("Panel configuration updated. Service might restart.", "success") # Alerts settings.cpu_limit = _parse_non_negative_int(request.form.get('cpu_limit', 90), default=90, minimum=1, maximum=100) settings.ram_limit = _parse_non_negative_int(request.form.get('ram_limit', 90), default=90, minimum=1, maximum=100) settings.disk_limit = _parse_non_negative_int(request.form.get('disk_limit', 90), default=90, minimum=1, maximum=100) # Instance Updates if request.form.get('auto_update') == 'on': settings.auto_update = True else: settings.auto_update = False db.session.commit() flash('Settings updated', 'success') return redirect(url_for('settings_page')) @app.route('/download_ca/') def download_ca(ca_id): if 'user_id' not in session: return redirect(url_for('login')) from models import CertificateAuthority ca = CertificateAuthority.query.get(ca_id) if not ca or not os.path.exists(ca.cert_path): flash('Certificate Authority not found.', 'error') return redirect(url_for('settings_page')) return send_file(ca.cert_path, as_attachment=True, download_name=f"{ca.name.replace(' ', '_')}.crt") @app.route('/check_updates', methods=['POST']) def check_updates(): if 'user_id' not in session: return redirect(url_for('login')) settings = Settings.query.first() has_update, new_version, err = update_manager.check_for_updates() settings.last_update_check = datetime.now() db.session.commit() if err: flash(f"Update Check Error: {err}", "error") elif has_update: flash(f"New Version Available: {new_version}. Install now available.", "info") session['available_update'] = new_version # Create alert and send email if not already alerted for this version msg = f"New version available: {new_version}. Current: {VERSION}" existing_alert = Alert.query.filter(Alert.subject == "Update Available", Alert.message.like(f"%{new_version}%")).first() if not existing_alert: admins = Admin.query.all() emails_to_notify = [a.email for a in admins if a.notify_update_available and a.email] if emails_to_notify: for email in emails_to_notify: monitor.send_alert(settings, "Update Available", msg, target_email=email) else: monitor.send_alert(settings, "Update Available", msg, disable_email=True) else: flash("You are using the latest version.", "success") session.pop('available_update', None) return redirect(url_for('settings_page')) @app.route('/start_install_update', methods=['POST']) def start_install_update(): if 'user_id' not in session: return {'error': 'Unauthorized'}, 401 # Determine path to helper script script_path = os.path.join(app.root_path, 'run_update_task.py') cmd = [sys.executable, script_path] task_id, err = task_manager.start_task(cmd) if not task_id: return {'error': err}, 500 return {'task_id': task_id} @app.route('/install_update', methods=['POST']) def install_update(): # Deprecated sync route, kept redirecting to settings to handle legacy calls gracefully return redirect(url_for('settings_page')) @app.route('/test_email', methods=['POST']) def test_email(): if 'user_id' not in session: return redirect(url_for('login')) user = Admin.query.get(session['user_id']) settings = Settings.query.first() target_email = user.email if user and user.email else (settings.alert_email if settings else None) if not settings or not settings.smtp_server or not target_email: flash("SMTP server or Profile/Alert Email is not configured.", "error") return redirect(url_for('settings_page')) try: import smtplib from email.mime.text import MIMEText msg = MIMEText("This is a test email from Limristem Web Control Panel.") msg['Subject'] = "[Limristem Web Alert] Test Email" msg['From'] = settings.smtp_user msg['To'] = target_email with smtplib.SMTP(settings.smtp_server, settings.smtp_port, timeout=10) as server: server.starttls() server.login(settings.smtp_user, settings.smtp_pass) server.send_message(msg) flash('Test email sent successfully!', 'success') except Exception as e: flash(f'Failed to send email. Raw error: {str(e)}', 'error') return redirect(url_for('settings_page')) @app.route('/test_webhook', methods=['POST']) def test_webhook(): if 'user_id' not in session: return redirect(url_for('login')) settings = Settings.query.first() webhook_url = (request.form.get('webhook_url') or '').strip() if webhook_url and not is_safe_external_url(webhook_url, block_internal=True): flash('Invalid webhook URL.', 'error') return redirect(url_for('settings_page')) webhook_headers = (request.form.get('webhook_headers') or '').strip() if webhook_headers: try: parsed_headers = json.loads(webhook_headers) if not isinstance(parsed_headers, dict): raise ValueError() except ValueError: flash('Webhook headers must be a valid JSON object.', 'error') return redirect(url_for('settings_page')) settings.webhook_url = webhook_url or None settings.webhook_headers = webhook_headers or None db.session.commit() try: monitor.send_alert(settings, "Test Webhook", "This is a test webhook from Limristem Web Control Panel.") flash('Test webhook sent (check response). Settings Saved.', 'success') except Exception as e: flash(f'Failed to send webhook: {e}', 'error') return redirect(url_for('settings_page')) @app.cli.command("show-admin") @click.option('--reveal', is_flag=True, help='Reset and reveal a new password') def show_admin(reveal): """Show Admin Username. Use --reveal to reset and show new password.""" # Note: CLI commands are already in app context, but explicit doesn't hurt if called differently admin = Admin.query.first() if admin: print(f"Username: {admin.username}") if reveal: new_pass = secrets.token_hex(8) admin.password = generate_password_hash(new_pass) db.session.commit() print(f"New Password: {new_pass}") print("WARNING: The password has been reset. Please save it now.") else: print("Password: (The password is hashed and cannot be retrieved)") print("Use 'flask show-admin --reveal' to reset and show a new password.") else: print("No admin user found.") return app if __name__ == '__main__': app = create_app() app.run(host='0.0.0.0', port=5000)