from flask import Flask, render_template, request, redirect, url_for, flash, session, Response, jsonify from flask_sqlalchemy import SQLAlchemy from werkzeug.security import generate_password_hash, check_password_hash import os, paramiko, threading, time, io, tempfile, csv from apscheduler.schedulers.background import BackgroundScheduler from datetime import datetime, timezone, timedelta from io import StringIO import socket import ipaddress import difflib from croniter import croniter from tzlocal import get_localzone from werkzeug.serving import WSGIRequestHandler WSGIRequestHandler.server_version = "" WSGIRequestHandler.sys_version = "" app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db' app.config['SECRET_KEY'] = 'supersecretkey' db = SQLAlchemy(app) # MODELE class User(db.Model): id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True, nullable=False) password = db.Column(db.String(200), nullable=False) hosts = db.relationship('Host', backref='user', lazy=True) hostfiles = db.relationship('HostFile', backref='user', lazy=True) settings = db.relationship('UserSettings', backref='user', uselist=False) class Host(db.Model): id = db.Column(db.Integer, primary_key=True) hostname = db.Column(db.String(255), nullable=False) username = db.Column(db.String(100), nullable=False) password = db.Column(db.String(200), nullable=False) type = db.Column(db.String(50), default='linux') # 'linux' albo 'mikrotik' auth_method = db.Column(db.String(20), default='password') private_key = db.Column(db.Text, nullable=True) key_passphrase = db.Column(db.String(200), nullable=True) port = db.Column(db.Integer, default=22) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) auto_deploy_enabled = db.Column(db.Boolean, default=True) auto_backup_enabled = db.Column(db.Boolean, default=True) preferred_hostfile_id = db.Column(db.Integer, db.ForeignKey('host_file.id'), nullable=True) preferred_hostfile = db.relationship('HostFile', foreign_keys=[preferred_hostfile_id]) use_daemon = db.Column(db.Boolean, default=False) daemon_url = db.Column(db.String(255), nullable=True) daemon_token = db.Column(db.String(255), nullable=True) disable_regex_deploy = db.Column(db.Boolean, default=False) disable_local_default = db.Column(db.Boolean, default=False) @property def resolved_hostname(self): try: return socket.gethostbyaddr(self.hostname)[0] except Exception: return self.hostname @property def resolved_daemon(self): if self.daemon_url: try: daemon_str = self.daemon_url.split("://")[-1] daemon_ip = daemon_str.split(":")[0] return socket.gethostbyaddr(daemon_ip)[0] except Exception: return daemon_ip return "" @property def raw_ip(self): if self.use_daemon and self.type == 'linux' and self.daemon_url: daemon_str = self.daemon_url.split("://")[-1] daemon_ip = daemon_str.split(":")[0] return daemon_ip return self.hostname class DeployLog(db.Model): id = db.Column(db.Integer, primary_key=True) timestamp = db.Column(db.DateTime, default=db.func.current_timestamp()) details = db.Column(db.Text, nullable=False) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) user = db.relationship('User', backref='deploy_logs', lazy=True) class HostFile(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) title = db.Column(db.String(100), nullable=False, default='Default Hosts') content = db.Column(db.Text, nullable=False) class UserSettings(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), unique=True, nullable=False) auto_deploy_enabled = db.Column(db.Boolean, default=False) #deploy_interval = db.Column(db.Integer, default=60) # interwał wdrożeń (minuty) #backup_interval = db.Column(db.Integer, default=60) # interwał backupów (minuty) deploy_cron = db.Column(db.String(100), default="12 12 * * *") backup_cron = db.Column(db.String(100), default="12 12 * * *") auto_backup_enabled = db.Column(db.Boolean, default=False) last_deploy_time = db.Column(db.DateTime, nullable=True) regex_deploy_enabled = db.Column(db.Boolean, default=True) backup_retention_days = db.Column(db.Integer, default=0) global_ssh_key = db.Column(db.Text, nullable=True) global_key_passphrase = db.Column(db.String(200), nullable=True) class Backup(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) host_id = db.Column(db.Integer, db.ForeignKey('host.id'), nullable=True) created_at = db.Column(db.DateTime, default=db.func.current_timestamp()) content = db.Column(db.Text, nullable=False) description = db.Column(db.String(255), nullable=True) host = db.relationship('Host', backref='backups', lazy=True) class RegexHostEntry(db.Model): __tablename__ = 'regex_host_entry' id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) cidr_range = db.Column(db.String(50), nullable=False) # np. '10.87.200.0/27' gateway_ip = db.Column(db.String(50), nullable=True) # np. '10.87.200.30' gateway_hostname = db.Column(db.String(255), nullable=True) # np. 'gw' domain_suffix = db.Column(db.String(255), nullable=False, default="guest.domain.com") host_prefix = db.Column(db.String(255), nullable=False, default="user") use_gateway_ip = db.Column(db.Boolean, default=False) comment = db.Column(db.String(255), nullable=True) user = db.relationship('User', backref='regex_entries') class HostFileVersion(db.Model): id = db.Column(db.Integer, primary_key=True) hostfile_id = db.Column(db.Integer, db.ForeignKey('host_file.id'), nullable=False) content = db.Column(db.Text, nullable=False) timestamp = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc)) hostfile = db.relationship('HostFile', backref=db.backref('versions', lazy=True)) class LocalDefaultEntry(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) ip_address = db.Column(db.String(50), nullable=False) hostname = db.Column(db.String(255), nullable=False) entry = db.Column(db.Text, nullable=False) # To pole już istnieje user = db.relationship('User', backref='local_defaults') # Funkcje pomocnicze def ensure_local_defaults(content, user_id): default_entries = LocalDefaultEntry.query.filter_by(user_id=user_id).all() required_lines = [] for entry in default_entries: if entry.ip_address and entry.hostname: required_lines.append(f"{entry.ip_address} {entry.hostname}".strip()) lines = [l.rstrip() for l in content.splitlines() if l.strip()] lines = [line for line in lines if not any(line.startswith(e.split()[0]) for e in required_lines)] lines = required_lines + lines final_content = "\n".join(lines) + "\n" return final_content def format_host(host): if not host: return "Unknown Host" resolved_name = None # Priorytet dla Linux Daemon if host.use_daemon and host.type == 'linux' and host.daemon_url: resolved_name = host.resolved_daemon or host.hostname # Dla standardowych hostów używamy resolved_hostname if not resolved_name: resolved_name = host.resolved_hostname or host.hostname # Jeśli resolved_name nadal jest IP, spróbuj rozwiązać przez DNS if resolved_name == host.raw_ip: try: resolved_name = socket.gethostbyaddr(host.raw_ip)[0] except (socket.herror, socket.gaierror): pass # Jeśli nie można rozwiązać, pozostaw IP return f"{resolved_name} ({host.raw_ip})" if resolved_name else f"Unknown ({host.raw_ip})" def wrap_content_with_comments(content): now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") header_comment = f"# Auto-hosts upload: {now_str}\n" footer_comment = f"\n# End of auto-hosts upload: {now_str}\n" return header_comment + content + footer_comment def open_ssh_connection(host_obj): ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) if host_obj.auth_method in ['ssh_key', 'global_key']: if host_obj.auth_method == 'ssh_key': key_str = host_obj.private_key key_passphrase = host_obj.key_passphrase if host_obj.key_passphrase else None else: # global_key # Pobieramy globalny klucz z ustawień użytkownika user_settings = UserSettings.query.filter_by(user_id=host_obj.user_id).first() if not user_settings or not user_settings.global_ssh_key: raise Exception("Globalny klucz SSH nie został ustawiony w ustawieniach.") key_str = user_settings.global_ssh_key key_passphrase = user_settings.global_key_passphrase if user_settings.global_key_passphrase else None key_file_obj = io.StringIO(key_str) try: pkey = paramiko.RSAKey.from_private_key(key_file_obj, password=key_passphrase) except paramiko.SSHException as e: raise Exception(f"Error reading private key: {str(e)}") ssh.connect( hostname=host_obj.hostname, port=host_obj.port, username=host_obj.username, pkey=pkey, timeout=10, banner_timeout=30 ) else: ssh.connect( hostname=host_obj.hostname, port=host_obj.port, username=host_obj.username, password=host_obj.password, timeout=10, banner_timeout=30 ) return ssh def get_statistics(user_id): user = db.session.get(User, user_id) host_count = Host.query.filter_by(user_id=user_id).count() logs = DeployLog.query.all() filtered_logs = [log for log in logs if f"for user {user_id}" in log.details or f"for {user.username}" in log.details] total_deployments = len(filtered_logs) successful_deployments = len([log for log in filtered_logs if "Updated" in log.details or "Deployed" in log.details]) failed_deployments = len([log for log in filtered_logs if "Failed" in log.details]) return { "host_count": host_count, "total_deployments": total_deployments, "successful_deployments": successful_deployments, "failed_deployments": failed_deployments } def automated_backup_for_host(host): try: if host.use_daemon and host.type == 'linux': import requests url = host.daemon_url.rstrip('/') + '/hosts' headers = {"Authorization": host.daemon_token} resp = requests.get(url, headers=headers, timeout=10, verify=False) if resp.status_code != 200: raise Exception(f"Daemon GET error: {resp.status_code} - {resp.text}") data = resp.json() content = data.get("hosts", "") # Używamy format_host, aby log był zgodny z deploy backup_info = f"[BACKUP] Automatic backup created for server {format_host(host)}" else: if host.type == 'mikrotik': ssh = open_ssh_connection(host) stdin, stdout, stderr = ssh.exec_command("/ip dns static export") content = stdout.read().decode('utf-8') ssh.close() backup_info = f"[BACKUP] Automatic backup created for server {format_host(host)}" else: ssh = open_ssh_connection(host) sftp = ssh.open_sftp() with sftp.open('/etc/hosts', 'r') as remote_file: content = remote_file.read().decode('utf-8') sftp.close() ssh.close() backup_info = f"[BACKUP] Automatic backup created for server {format_host(host)}" backup = Backup( user_id=host.user_id, host_id=host.id, content=content, description=f'Backup from server {format_host(host)} at {datetime.now(timezone.utc).isoformat()}' ) db.session.add(backup) db.session.commit() log_entry = DeployLog(details=backup_info, user_id=host.user_id) db.session.add(log_entry) db.session.commit() print(f'Automated backup for server {format_host(host)} created successfully.') except Exception as e: print(f'Error creating automated backup for server {format_host(host)}: {str(e)}') def automated_backups(): with app.app_context(): now = datetime.now(timezone.utc) hosts = Host.query.all() for host in hosts: # Dodaj warunek: backup dla danego hosta ma być wykonywany tylko, jeśli jest włączony if not host.auto_backup_enabled: continue settings = UserSettings.query.filter_by(user_id=host.user_id).first() if not settings or not settings.auto_backup_enabled or not settings.backup_cron: continue # Pobieramy ostatni backup dla hosta last_backup = Backup.query.filter_by(user_id=host.user_id, host_id=host.id)\ .order_by(Backup.created_at.desc()).first() if last_backup: base_time = last_backup.created_at if base_time.tzinfo is None: base_time = base_time.replace(tzinfo=timezone.utc) else: base_time = datetime.now(timezone.utc) - timedelta(minutes=1) cron = croniter(settings.backup_cron, base_time) next_backup_time = cron.get_next(datetime) if now >= next_backup_time: automated_backup_for_host(host) db.session.commit() def wrap_content_with_comments(content): now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") header_comment = f"# Auto-hosts upload: {now_str}\n" footer_comment = f"\n# End of auto-hosts upload: {now_str}\n" return header_comment + content + footer_comment def wrap_mikrotik_content(content): now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d_%H:%M:%S_UTC") header = f"127.0.0.99 #Auto-hosts_upload:{now_str}" footer = f"127.0.0.199 #End_of_auto-hosts_upload:{now_str}" return header + "\n" + content + "\n" + footer def clear_linux(host, content): """Zastępuje /etc/hosts na hoście Linux zawartością `content`.""" ssh = open_ssh_connection(host) import tempfile, os with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmpf: tmpf.write(content) tmp_file_path = tmpf.name sftp = ssh.open_sftp() sftp.put(tmp_file_path, '/etc/hosts') sftp.close() ssh.close() os.remove(tmp_file_path) def clear_mikrotik(host): ssh = open_ssh_connection(host) ssh.exec_command("/ip dns static remove [find]") ssh.close() def generate_regex_hosts(user_id): entries = RegexHostEntry.query.filter_by(user_id=user_id).all() lines = [] for entry in entries: try: network = ipaddress.ip_network(entry.cidr_range, strict=False) all_hosts = list(network.hosts()) # IP w puli host_index = 1 for ip_addr in all_hosts: if (entry.use_gateway_ip and entry.gateway_ip and str(ip_addr) == entry.gateway_ip): if entry.gateway_hostname: hostname = f"{entry.gateway_hostname}.{entry.domain_suffix}" else: hostname = f"gw.{entry.domain_suffix}" lines.append(f"{ip_addr} {hostname}") else: hostname = f"{entry.host_prefix}{host_index}.{entry.domain_suffix}" lines.append(f"{ip_addr} {hostname}") host_index += 1 except ValueError: pass return "\n".join(lines) + "\n" if lines else "" def cleanup_old_backups(): with app.app_context(): all_settings = UserSettings.query.all() for setting in all_settings: days = setting.backup_retention_days or 0 if days > 0: cutoff_date = datetime.now(timezone.utc) - timedelta(days=days) old_backups = Backup.query.filter( Backup.user_id == setting.user_id, Backup.created_at < cutoff_date ).all() for b in old_backups: db.session.delete(b) db.session.commit() def format_host(host): if host.use_daemon and host.type == 'linux' and host.daemon_url: resolved_name = host.resolved_daemon or host.hostname else: resolved_name = host.resolved_hostname or host.hostname return f"{resolved_name} ({host.raw_ip})" def remove_local_defaults(content, user_id): local_defaults = LocalDefaultEntry.query.filter_by(user_id=user_id).all() lines = content.splitlines() cleaned_lines = [] for line in lines: stripped_line = line.strip() if not any(stripped_line.startswith(f"{entry.ip_address} {entry.hostname}") for entry in local_defaults): cleaned_lines.append(stripped_line) return "\n".join(cleaned_lines) + "\n" # ------------------- # LOGOWANIE, REJESTRACJA, ZMIANA HASŁA # ------------------- @app.route('/') def index(): if 'user_id' in session: return redirect(url_for('dashboard')) return redirect(url_for('login')) @app.route('/login', methods=['GET', 'POST']) def login(): if request.method == 'POST': username = request.form['username'] password = request.form['password'] user = User.query.filter_by(username=username).first() if user and check_password_hash(user.password, password): session['user_id'] = user.id flash('Login successful', 'success') return redirect(url_for('dashboard')) flash('Invalid credentials', 'danger') return render_template('login.html') @app.route('/register', methods=['GET', 'POST']) def register(): if request.method == 'POST': username = request.form['username'] password_hash = generate_password_hash(request.form['password']) user = User(username=username, password=password_hash) db.session.add(user) db.session.commit() flash('Registration successful. Please log in.', 'success') return redirect(url_for('login')) return render_template('register.html') @app.route('/logout') def logout(): session.pop('user_id', None) flash('Logged out', 'info') return redirect(url_for('login')) @app.route('/change-password', methods=['GET', 'POST']) def change_password(): if 'user_id' not in session: return redirect(url_for('login')) if request.method == 'POST': user = db.session.get(User, session['user_id']) new_password = generate_password_hash(request.form['password']) user.password = new_password db.session.commit() flash('Password changed successfully', 'success') return redirect(url_for('change_password')) return render_template('change_password.html') # ------------------- # ZARZĄDZANIE SERWERAMI # ------------------- @app.route('/add_server', methods=['GET', 'POST']) def add_server(): if 'user_id' not in session: return redirect(url_for('login')) # Pobieramy wszystkie pliki hosts dla użytkownika (do wyboru preferowanego) user_hostfiles = HostFile.query.filter_by(user_id=session['user_id']).all() if request.method == 'POST': hostname = request.form['hostname'] username = request.form['username'] password_val = request.form['password'] host_type = request.form.get('host_type', 'linux') auth_method = request.form.get('auth_method', 'password') private_key = request.form.get('private_key', '').strip() key_passphrase = request.form.get('key_passphrase', '').strip() port_str = request.form.get('port', '22') try: port = int(port_str) except ValueError: port = 22 # Używamy danych dla demona, jeśli checkbox zaznaczony use_daemon = bool(request.form.get('use_daemon')) daemon_url = request.form.get('daemon_url', '').strip() daemon_token = request.form.get('daemon_token', '').strip() # Dla metod 'ssh_key' i 'global_key' dane są rozróżniane: if auth_method == 'ssh_key': stored_key = private_key if private_key else None stored_passphrase = key_passphrase if key_passphrase else None elif auth_method == 'global_key': # W przypadku global_key dane lokalne nie są zapisywane – będą pobierane z ustawień użytkownika stored_key = None stored_passphrase = None else: stored_key = None stored_passphrase = None # Obsługa preferowanego pliku hosts preferred_file_id_str = request.form.get('preferred_hostfile_id', '').strip() if preferred_file_id_str == '': chosen_file_id = None else: try: chosen_file_id = int(preferred_file_id_str) except ValueError: chosen_file_id = None host = Host( hostname=hostname, username=username, password=password_val, type=host_type, auth_method=auth_method, private_key=stored_key, key_passphrase=stored_passphrase, port=port, user_id=session['user_id'], use_daemon=use_daemon if host_type == 'linux' else False, daemon_url=daemon_url if (use_daemon and host_type == 'linux') else None, daemon_token=daemon_token if (use_daemon and host_type == 'linux') else None, preferred_hostfile_id=chosen_file_id ) db.session.add(host) db.session.commit() flash('Host added successfully', 'success') return redirect(url_for('server_list')) return render_template('add_server.html', user_hostfiles=user_hostfiles) @app.route('/delete-server/') def delete_server(id): if 'user_id' not in session: return redirect(url_for('login')) host = db.session.get(Host, id) if host and host.user_id == session['user_id']: db.session.delete(host) db.session.commit() flash('Host deleted', 'info') else: flash('Host not found or unauthorized', 'danger') return redirect(url_for('server_list')) @app.route('/server-list', methods=['GET']) def server_list(): if 'user_id' not in session: return redirect(url_for('login')) hosts = Host.query.filter_by(user_id=session['user_id']).all() return render_template('server_list.html', hosts=hosts) @app.route('/edit-server/', methods=['GET', 'POST']) def edit_server(id): if 'user_id' not in session: return redirect(url_for('login')) host = db.session.get(Host, id) if not host or host.user_id != session['user_id']: flash('Server not found or unauthorized', 'danger') return redirect(url_for('server_list')) # Pobieramy listę plików hosts dla użytkownika (do wyboru preferowanego) user_hostfiles = HostFile.query.filter_by(user_id=session['user_id']).all() if request.method == 'POST': host.hostname = request.form['hostname'] host.username = request.form['username'] new_password = request.form['password'] or '' if new_password: host.password = new_password port_str = request.form.get('port', '22') try: host.port = int(port_str) except ValueError: host.port = 22 host.type = request.form.get('host_type', 'linux') host.auth_method = request.form.get('auth_method', 'password') if host.auth_method == 'ssh_key': new_private_key = request.form.get('private_key', '').strip() new_passphrase = request.form.get('key_passphrase', '').strip() if new_private_key: host.private_key = new_private_key if new_passphrase: host.key_passphrase = new_passphrase elif host.auth_method == 'global_key': # Dla global_key wyczyścimy lokalne pola – dane będą pobierane z ustawień host.private_key = None host.key_passphrase = None use_daemon = bool(request.form.get('use_daemon')) daemon_url = request.form.get('daemon_url', '').strip() daemon_token = request.form.get('daemon_token', '').strip() if host.type == 'linux' and use_daemon: host.use_daemon = True host.daemon_url = daemon_url host.daemon_token = daemon_token else: host.use_daemon = False host.daemon_url = None host.daemon_token = None preferred_file_id_str = request.form.get('preferred_hostfile_id', '').strip() if preferred_file_id_str == '': host.preferred_hostfile_id = None else: try: host.preferred_hostfile_id = int(preferred_file_id_str) except ValueError: host.preferred_hostfile_id = None db.session.commit() flash('Server updated successfully', 'success') return redirect(url_for('server_list')) return render_template('edit_server.html', host=host, user_hostfiles=user_hostfiles) # ------------------- # TESTOWANIE POŁĄCZENIA SSH DLA HOSTA # ------------------- @app.route('/test-server-connection/', methods=['GET']) def test_server_connection(id): if 'user_id' not in session: return redirect(url_for('login')) host = db.session.get(Host, id) if not host or host.user_id != session['user_id']: flash('Host not found or unauthorized', 'danger') return redirect(url_for('server_list')) try: if host.use_daemon and host.type == 'linux': import requests headers = {"Authorization": host.daemon_token} health_url = host.daemon_url.rstrip('/') + '/health' resp = requests.get(health_url, headers=headers, verify=False, timeout=5) if resp.status_code == 200: flash(f'Demon connection successful (health OK) for {format_host(host)}', 'success') else: raise Exception(f"Demon health check returned {resp.status_code}") sysinfo_url = host.daemon_url.rstrip('/') + '/system-info' sysinfo_resp = requests.get(sysinfo_url, headers=headers, verify=False, timeout=5) if sysinfo_resp.status_code == 200: info = sysinfo_resp.json() msg = (f"System-info for {format_host(host)}: " f"CPU={info.get('cpu_percent')}%, " f"MEM={info.get('memory_percent')}%, " f"DISK={info.get('disk_percent')}%, " f"UPTIME={info.get('uptime_seconds')}s") flash(msg, 'info') else: raise Exception(f"Demon system-info returned {sysinfo_resp.status_code}") else: ssh = open_ssh_connection(host) ssh.close() flash(f'SSH connection to {format_host(host)} successful.', 'success') except Exception as e: flash(f'Connection failed for {format_host(host)}: {str(e)}', 'danger') return redirect(url_for('server_list')) # ------------------- # ROUTE: CZYSZCZENIE HOSTS - CAŁA GRUPA # ------------------- @app.route('/clear-server', methods=['GET', 'POST']) def clear_server(): if 'user_id' not in session: return redirect(url_for('login')) hosts = Host.query.filter_by(user_id=session['user_id']).all() return render_template('clear_servers.html', hosts=hosts, format_host=format_host) @app.route('/clear-single-server/', methods=['POST']) def clear_single_server(host_id): if 'user_id' not in session: return redirect(url_for('login')) host = db.session.get(Host, host_id) if not host or host.user_id != session['user_id']: flash('Host not found or unauthorized', 'danger') return redirect(url_for('clear_server')) default_content = ensure_local_defaults("", session['user_id']) try: host_name = format_host(host) if host.use_daemon and host.type == 'linux': import requests url = host.daemon_url.rstrip('/') + '/hosts' headers = {"Authorization": host.daemon_token} resp = requests.post(url, json={"hosts": default_content}, headers=headers, verify=False, timeout=10) if resp.status_code != 200: raise Exception(f"Daemon update error: {resp.status_code} - {resp.text}") elif host.type == 'mikrotik': clear_mikrotik(host) else: clear_linux(host, default_content) flash(f'Cleared host: {host_name}', 'success') except Exception as e: flash(f'Error clearing host {host_name}: {str(e)}', 'danger') return redirect(url_for('clear_server')) @app.route('/clear-all-server', methods=['GET', 'POST']) def clear_all_server(): if 'user_id' not in session: return redirect(url_for('login')) hosts = Host.query.filter_by(user_id=session['user_id']).all() if request.method == 'POST': linux_clear = request.form.get('linux') mikrotik_clear = request.form.get('mikrotik') default_content = ensure_local_defaults("", session['user_id']) for h in hosts: try: host_name = format_host(h) if h else "Unknown Host" print(f"DEBUG: Czyszczenie hosta {host_name}") if h.type == 'linux' and linux_clear: if h.use_daemon: import requests url = h.daemon_url.rstrip('/') + '/hosts' headers = {"Authorization": h.daemon_token} resp = requests.post(url, json={"hosts": default_content}, headers=headers, verify=False, timeout=10) if resp.status_code != 200: raise Exception(f"Daemon update error: {resp.status_code} - {resp.text}") else: clear_linux(h, default_content) flash(f'Cleared Linux host: {host_name}', 'success') elif h.type == 'mikrotik' and mikrotik_clear: clear_mikrotik(h) flash(f'Cleared Mikrotik host: {host_name}', 'success') except Exception as e: flash(f'Error clearing host {host_name}: {str(e)}', 'danger') return redirect(url_for('clear_all_server')) return render_template('clear_servers.html', hosts=hosts, format_host=format_host) # ------------------- # ZARZĄDZANIE PLIKAMI HOSTS (WIELOKROTNE PLIKI) # ------------------- @app.route('/hosts-files') def list_hosts_files(): if 'user_id' not in session: return redirect(url_for('login')) files = HostFile.query.filter_by(user_id=session['user_id']).all() return render_template('hosts_files.html', files=files) @app.route('/hosts-files/new', methods=['GET', 'POST']) def new_hosts_file(): if 'user_id' not in session: return redirect(url_for('login')) if request.method == 'POST': title = request.form['title'] content = request.form['content'] new_file = HostFile(user_id=session['user_id'], title=title, content=content) db.session.add(new_file) db.session.commit() flash('Hosts file created', 'success') return redirect(url_for('list_hosts_files')) return render_template('new_edit_hosts_file.html', file=None) @app.route('/hosts-files//edit', methods=['GET', 'POST']) def edit_hosts_file(file_id): if 'user_id' not in session: return redirect(url_for('login')) file = db.session.get(HostFile, file_id) if not file or file.user_id != session['user_id']: flash('File not found or unauthorized', 'danger') return redirect(url_for('list_hosts_files')) if request.method == 'POST': new_title = request.form['title'] new_content = request.form['content'] if file.content != new_content: version = HostFileVersion(hostfile_id=file.id, content=file.content) db.session.add(version) file.title = new_title file.content = new_content db.session.commit() flash('Hosts file updated and previous version saved.', 'success') return redirect(url_for('list_hosts_files')) return render_template('new_edit_hosts_file.html', file=file) @app.route('/hosts-files//delete', methods=['GET']) def delete_hosts_file(file_id): if 'user_id' not in session: return redirect(url_for('login')) file = db.session.get(HostFile, file_id) if not file or file.user_id != session['user_id']: flash('File not found or unauthorized', 'danger') else: db.session.delete(file) db.session.commit() flash('Hosts file deleted', 'info') return redirect(url_for('list_hosts_files')) # ------------------- # WDROŻENIE WYBRANEGO PLIKU HOSTS NA WYBRANE SERWERY # ------------------- @app.route('/deploy-hosts-file/', methods=['GET', 'POST']) def deploy_hosts_file(file_id): if 'user_id' not in session: return redirect(url_for('login')) file = db.session.get(HostFile, file_id) if not file or file.user_id != session['user_id']: flash('File not found or unauthorized', 'danger') return redirect(url_for('list_hosts_files')) hosts = Host.query.filter_by(user_id=session['user_id']).all() # Dla hostów korzystających z demona – obliczamy IP oraz rozwiązaną nazwę for host in hosts: if host.use_daemon and host.type == 'linux' and host.daemon_url: daemon_str = host.daemon_url.split("://")[-1] daemon_ip = daemon_str.split(":")[0] host.daemon_ip = daemon_ip try: resolved_daemon = socket.gethostbyaddr(daemon_ip)[0] except Exception: resolved_daemon = daemon_ip host._resolved_daemon_local = resolved_daemon if request.method == 'POST': selected_host_ids = request.form.getlist('hosts') for host in hosts: if str(host.id) in selected_host_ids: try: adjusted_content = ensure_local_defaults(file.content) wrapped_content = wrap_content_with_comments(adjusted_content) if host.use_daemon and host.type == 'linux': import requests url = host.daemon_url.rstrip('/') + '/hosts' headers = {"Authorization": host.daemon_token} resp = requests.post(url, json={"hosts": wrapped_content}, headers=headers, timeout=10, verify=False) if resp.status_code != 200: raise Exception(f"Daemon POST error: {resp.status_code} - {resp.text}") log_details = f'[LINUX/DAEMON] Updated {format_host(host)}' db.session.add(DeployLog(details=log_details, user_id=session['user_id'])) elif host.type == 'mikrotik': wrapped_mikro = wrap_mikrotik_content(file.content) deploy_mikrotik(host, wrapped_mikro) log_details = f'[MIKROTIK] Updated {format_host(host)}' db.session.add(DeployLog(details=log_details, user_id=session['user_id'])) else: ssh = open_ssh_connection(host) with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmpf: tmpf.write(wrapped_content) tmp_file_path = tmpf.name sftp = ssh.open_sftp() sftp.put(tmp_file_path, '/etc/hosts') sftp.close() ssh.close() os.remove(tmp_file_path) log_details = f'[LINUX] Updated {format_host(host)}' db.session.add(DeployLog(details=log_details, user_id=session['user_id'])) db.session.commit() flash(f'Deployed file "{file.title}" to {format_host(host)}', 'success') except Exception as e: flash(f'Error deploying file "{file.title}" to {format_host(host)}: {str(e)}', 'danger') return redirect(url_for('list_hosts_files')) return render_template('deploy_hosts_file.html', file=file, hosts=hosts) # ------------------- # BACKUP # ------------------- @app.route('/server-backup/', methods=['GET']) def server_backup(host_id): if 'user_id' not in session: return redirect(url_for('login')) host = db.session.get(Host, host_id) if not host or host.user_id != session['user_id']: flash('Host not found or unauthorized', 'danger') return redirect(url_for('server_list')) try: if host.use_daemon and host.type == 'linux': import requests url = host.daemon_url.rstrip('/') + '/hosts' headers = {"Authorization": host.daemon_token} resp = requests.get(url, headers=headers, timeout=10, verify=False) if resp.status_code != 200: raise Exception(f"Daemon GET error: {resp.status_code} - {resp.text}") data = resp.json() content = data.get("hosts", "") description = f'Backup from server {format_host(host)} at {datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")}' elif host.type == 'mikrotik': ssh = open_ssh_connection(host) stdin, stdout, stderr = ssh.exec_command("/ip dns static export") content = stdout.read().decode('utf-8') ssh.close() description = f'Backup from server {format_host(host)}' else: ssh = open_ssh_connection(host) sftp = ssh.open_sftp() with sftp.open('/etc/hosts', 'r') as remote_file: content = remote_file.read().decode('utf-8') sftp.close() ssh.close() description = f'Backup from server {format_host(host)}' backup = Backup( user_id=session['user_id'], host_id=host.id, content=content, description=description ) db.session.add(backup) db.session.commit() flash(f'Backup for host {format_host(host)} created successfully.', 'success') except Exception as e: flash(f'Error creating backup for host {format_host(host)}: {str(e)}', 'danger') return redirect(url_for('server_list')) @app.route('/backups') def backups(): if 'user_id' not in session: return redirect(url_for('login')) backups = Backup.query.filter_by(user_id=session['user_id']).order_by(Backup.created_at.desc()).all() return render_template('backups.html', backups=backups) @app.route('/restore-backup/', methods=['GET']) def restore_backup(backup_id): if 'user_id' not in session: return redirect(url_for('login')) backup = db.session.get(Backup, backup_id) if not backup or backup.user_id != session['user_id']: flash('Backup not found or unauthorized', 'danger') return redirect(url_for('backups')) if backup.host_id: host = db.session.get(Host, backup.host_id) if not host or host.user_id != session['user_id']: flash('Associated host not found or unauthorized', 'danger') return redirect(url_for('backups')) try: if host.type == 'mikrotik': deploy_mikrotik(host, backup.content) flash(f'Backup restored to {format_host(host)} successfully.', 'success') else: ssh = open_ssh_connection(host) sftp = ssh.open_sftp() with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmpf: tmpf.write(backup.content) tmp_file_path = tmpf.name sftp.put(tmp_file_path, '/etc/hosts') sftp.close() ssh.close() os.remove(tmp_file_path) flash(f'Backup restored to {format_host(host)} successfully.', 'success') except Exception as e: flash(f'Error restoring backup to {format_host(host)}: {str(e)}', 'danger') else: hostfile = HostFile.query.filter_by(user_id=session['user_id'], title="Default Hosts").first() if not hostfile: hostfile = HostFile(user_id=session['user_id'], title="Default Hosts", content=backup.content) db.session.add(hostfile) else: hostfile.content = backup.content db.session.commit() deploy_user(session['user_id']) flash('Backup restored to default configuration and deployed successfully.', 'success') return redirect(url_for('backups')) @app.route('/view-backup/', methods=['GET']) def view_backup(backup_id): if 'user_id' not in session: return redirect(url_for('login')) backup = db.session.get(Backup, backup_id) if not backup or backup.user_id != session['user_id']: flash('Backup not found or unauthorized', 'danger') return redirect(url_for('backups')) host = None if backup.host_id: host = db.session.get(Host, backup.host_id) return render_template('view_backup.html', backup=backup, host=host) @app.route('/backup-all', methods=['GET']) def backup_all(): if 'user_id' not in session: return redirect(url_for('login')) user_id = session['user_id'] hosts = Host.query.filter_by(user_id=user_id).all() for host in hosts: try: if host.use_daemon and host.type == 'linux': import requests url = host.daemon_url.rstrip('/') + '/hosts' headers = {"Authorization": host.daemon_token} resp = requests.get(url, headers=headers, timeout=10, verify=False) if resp.status_code != 200: raise Exception(f"Daemon GET error: {resp.status_code} - {resp.text}") data = resp.json() content = data.get("hosts", "") description = f'Backup from server {format_host(host)}' elif host.type == 'mikrotik': ssh = open_ssh_connection(host) stdin, stdout, stderr = ssh.exec_command("/ip dns static export") content = stdout.read().decode('utf-8') ssh.close() description = f'Backup from server {format_host(host)}' else: ssh = open_ssh_connection(host) sftp = ssh.open_sftp() with sftp.open('/etc/hosts', 'r') as remote_file: content = remote_file.read().decode('utf-8') sftp.close() ssh.close() description = f'Backup from server {format_host(host)}' backup = Backup( user_id=user_id, host_id=host.id, content=content, description=description ) db.session.add(backup) db.session.commit() except Exception as e: flash(f'Error creating backup for {format_host(host)}: {str(e)}', 'danger') flash('Backup for all hosts created successfully.', 'success') return redirect(url_for('backups')) # ------------------- # IMPORT/EXPORT HOSTÓW # ------------------- @app.route('/export-servers-to-csv', methods=['GET']) def export_servers_to_csv(): if 'user_id' not in session: return redirect(url_for('login')) user_id = session['user_id'] hosts = Host.query.filter_by(user_id=user_id).all() si = StringIO() cw = csv.writer(si) # Dodajemy wszystkie istotne pola cw.writerow([ 'id', 'hostname', 'username', 'password', 'port', 'type', 'auth_method', 'private_key', 'key_passphrase', 'auto_deploy_enabled', 'auto_backup_enabled', 'preferred_hostfile_id', 'use_daemon', 'daemon_url', 'daemon_token' ]) for host in hosts: cw.writerow([ host.id, host.hostname, host.username, host.password, host.port, host.type, host.auth_method, host.private_key or '', host.key_passphrase or '', host.auto_deploy_enabled, host.auto_backup_enabled, host.preferred_hostfile_id if host.preferred_hostfile_id is not None else '', host.use_daemon, host.daemon_url or '', host.daemon_token or '' ]) output = si.getvalue() return Response(output, mimetype="text/csv", headers={"Content-Disposition": "attachment;filename=servers_full.csv"}) @app.route('/import-servers', methods=['GET', 'POST']) def import_servers(): if 'user_id' not in session: return redirect(url_for('login')) if request.method == 'POST': file = request.files.get('file') if not file: flash('No file uploaded', 'danger') return redirect(url_for('import_servers')) stream = StringIO(file.stream.read().decode("UTF8"), newline=None) csv_input = csv.reader(stream) header = next(csv_input) # zakładamy, że pierwszy wiersz zawiera nagłówki for row in csv_input: # Sprawdzamy, czy wiersz zawiera wszystkie wymagane kolumny (w tym przypadku 15) if len(row) < 15: continue # Rozpakowywanie wiersza zgodnie z kolejnością kolumn w eksporcie (_, hostname, username, password_val, port_str, host_type, auth_method, private_key, key_passphrase, auto_deploy_enabled, auto_backup_enabled, preferred_hostfile_id, use_daemon, daemon_url, daemon_token) = row try: port = int(port_str) except ValueError: port = 22 # Konwersja wartości logicznych auto_deploy_enabled = auto_deploy_enabled.lower() in ['true', '1', 'yes'] auto_backup_enabled = auto_backup_enabled.lower() in ['true', '1', 'yes'] use_daemon = use_daemon.lower() in ['true', '1', 'yes'] try: preferred_hostfile_id = int(preferred_hostfile_id) if preferred_hostfile_id else None except ValueError: preferred_hostfile_id = None host = Host( hostname=hostname, username=username, password=password_val, port=port, type=host_type, auth_method=auth_method, private_key=private_key if private_key else None, key_passphrase=key_passphrase if key_passphrase else None, auto_deploy_enabled=auto_deploy_enabled, auto_backup_enabled=auto_backup_enabled, preferred_hostfile_id=preferred_hostfile_id, use_daemon=use_daemon, daemon_url=daemon_url if daemon_url else None, daemon_token=daemon_token if daemon_token else None, user_id=session['user_id'] ) db.session.add(host) db.session.commit() flash('Hosts imported successfully', 'success') return redirect(url_for('server_list')) return render_template('import_servers.html') @app.route('/clear-host/', methods=['GET']) def clear_host(id): if 'user_id' not in session: return redirect(url_for('login')) host = db.session.get(Host, id) if not host or host.user_id != session['user_id']: flash('Host not found or unauthorized', 'danger') return redirect(url_for('server_list')) try: if host.type == 'linux': default_content = ensure_local_defaults("") clear_linux(host, default_content) elif host.type == 'mikrotik': clear_mikrotik(host) flash(f'Cleared host: {host.hostname}', 'success') except Exception as e: flash(f'Error clearing host {host.hostname}: {str(e)}', 'danger') return redirect(url_for('server_list')) # ------------------- # STRONA USTAWIEŃ (SETTINGS) # ------------------- @app.route('/settings', methods=['GET', 'POST']) def settings(): if 'user_id' not in session: return redirect(url_for('login')) user_id = session['user_id'] user_settings = UserSettings.query.filter_by(user_id=user_id).first() if not user_settings: user_settings = UserSettings(user_id=user_id) db.session.add(user_settings) db.session.commit() if request.method == 'POST': auto_deploy = request.form.get('auto_deploy') deploy_cron = request.form.get('deploy_cron') auto_backup = request.form.get('auto_backup') backup_cron = request.form.get('backup_cron') enable_regex_entries = request.form.get('enable_regex_entries') retention_val = request.form.get('backup_retention_days', '0') # Pobierz wartości globalnego klucza SSH z formularza global_ssh_key = request.form.get('global_ssh_key') global_key_passphrase = request.form.get('global_key_passphrase') # Walidacja wyrażeń cron przy pomocy croniter try: croniter(deploy_cron) except Exception as e: flash("Błędne wyrażenie cron dla deploy: " + str(e), "danger") return redirect(url_for('settings')) try: croniter(backup_cron) except Exception as e: flash("Błędne wyrażenie cron dla backup: " + str(e), "danger") return redirect(url_for('settings')) user_settings.auto_deploy_enabled = bool(auto_deploy) user_settings.auto_backup_enabled = bool(auto_backup) user_settings.deploy_cron = deploy_cron if deploy_cron else "12 12 * * *" user_settings.backup_cron = backup_cron if backup_cron else "12 12 * * *" user_settings.regex_deploy_enabled = bool(enable_regex_entries) try: user_settings.backup_retention_days = int(retention_val) except ValueError: user_settings.backup_retention_days = 0 # Zapis globalnego klucza SSH i passphrase user_settings.global_ssh_key = global_ssh_key user_settings.global_key_passphrase = global_key_passphrase db.session.commit() flash('Settings updated', 'success') return redirect(url_for('settings')) return render_template('settings.html', settings=user_settings) @app.route('/delete-backup/', methods=['POST']) def delete_backup(backup_id): if 'user_id' not in session: return redirect(url_for('login')) backup = Backup.query.get(backup_id) if not backup or backup.user_id != session['user_id']: flash('Backup not found or unauthorized', 'danger') return redirect(url_for('backups')) db.session.delete(backup) db.session.commit() flash('Backup deleted successfully', 'info') return redirect(url_for('backups')) # ------------------- # DEPLOYMENT DOMYŚLNY DLA UŻYTKOWNIKA # ------------------- def deploy_user(user_id): user_settings = UserSettings.query.filter_by(user_id=user_id).first() default_file = HostFile.query.filter_by(user_id=user_id, title="Default Hosts").first() if not default_file: return regex_lines = "" if user_settings and user_settings.regex_deploy_enabled: regex_lines = generate_regex_hosts(user_id) hosts = Host.query.filter_by(user_id=user_id).all() for h in hosts: if not h.auto_deploy_enabled: continue chosen_file = HostFile.query.filter_by(id=h.preferred_hostfile_id, user_id=user_id).first() if h.preferred_hostfile_id else default_file final_content = chosen_file.content.strip() if h.disable_local_default: final_content = remove_local_defaults(final_content, user_id) else: if not h.disable_regex_deploy and regex_lines.strip(): final_content = regex_lines + "\n" + final_content final_content = ensure_local_defaults(final_content, user_id) try: host_name = format_host(h) if h else "Unknown Host" if h.type == 'mikrotik': wrapped_content = wrap_mikrotik_content(final_content) deploy_mikrotik(h, wrapped_content) log_details = f'[MIKROTIK] Updated {host_name} for user {user_id}' elif h.use_daemon and h.type == 'linux': import requests wrapped_content = wrap_content_with_comments(final_content) url = h.daemon_url.rstrip('/') + '/hosts' headers = {"Authorization": h.daemon_token} resp = requests.post(url, json={"hosts": wrapped_content}, headers=headers, timeout=10, verify=False) if resp.status_code != 200: raise Exception(f"Daemon POST error: {resp.status_code} - {resp.text}") log_details = f'[LINUX/DAEMON] Updated {host_name} for user {user_id}' else: ssh = open_ssh_connection(h) wrapped_content = wrap_content_with_comments(final_content) with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmpf: tmpf.write(wrapped_content) tmp_file_path = tmpf.name sftp = ssh.open_sftp() sftp.put(tmp_file_path, '/etc/hosts') sftp.close() ssh.close() os.remove(tmp_file_path) log_details = f'[LINUX] Updated {host_name} for user {user_id}' db.session.add(DeployLog(details=log_details, user_id=user_id)) db.session.commit() except Exception as e: error_log = f'Failed to update {host_name}: {str(e)} for user {user_id}' db.session.add(DeployLog(details=error_log, user_id=user_id)) db.session.commit() def deploy_mikrotik(host, hosts_content): ssh = open_ssh_connection(host) stdin, stdout, stderr = ssh.exec_command("/ip dns static export") exported = stdout.read().decode('utf-8').splitlines() existing_dns = {} # Parsujemy aktualne wpisy z urządzenia (format: "add address=... name=...") for line in exported: line = line.strip() if not line.startswith('add '): continue line = line[4:].strip() # usuwamy "add " parts = line.split() address_val = None name_val = None for part in parts: if part.startswith('address='): address_val = part.replace('address=', '') elif part.startswith('name='): name_val = part.replace('name=', '') if address_val and name_val: existing_dns[name_val] = address_val desired_dns = {} # Parsujemy zawartość backupu/wdrażaną treść – obsługujemy dwa formaty: # 1. Format z tokenami ("add address=... name=...") # 2. Prosty format ("ip hostname") for line in hosts_content.splitlines(): line = line.strip() if not line: continue # Pomijamy linie nagłówkowe/stopowe if line.startswith('#') or 'Auto-hosts_upload:' in line or 'End_of_auto-hosts_upload:' in line: continue if "address=" in line and "name=" in line: # Format: "add address=... name=..." # Jeśli linia zaczyna się od "add ", usuńmy go, choć może już być usunięty if line.startswith('add '): line = line[4:].strip() parts = line.split() address_val = None name_val = None for part in parts: if part.startswith('address='): address_val = part.replace('address=', '') elif part.startswith('name='): name_val = part.replace('name=', '') if address_val and name_val: desired_dns[name_val] = address_val else: # Zakładamy prosty format: "ip hostname" (przynajmniej dwa tokeny) parts = line.split() if len(parts) >= 2: ip = parts[0] hostname = parts[1] desired_dns[hostname] = ip # Dodajemy lub aktualizujemy wpisy, a zbędne usuwamy for name_val, ip_val in desired_dns.items(): if name_val not in existing_dns: add_cmd = f"/ip dns static add address={ip_val} name={name_val}" ssh.exec_command(add_cmd) else: current_ip = existing_dns[name_val] if current_ip != ip_val: remove_cmd = f"/ip dns static remove [find where name={name_val}]" ssh.exec_command(remove_cmd) add_cmd = f"/ip dns static add address={ip_val} name={name_val}" ssh.exec_command(add_cmd) for existing_name in existing_dns: if existing_name not in desired_dns: remove_cmd = f"/ip dns static remove [find where name={existing_name}]" ssh.exec_command(remove_cmd) ssh.close() @app.route('/deploy-now') def deploy_now(): if 'user_id' not in session: return redirect(url_for('login')) deploy_user(session['user_id']) flash('Deployment complete', 'success') return redirect(url_for('dashboard')) # ------------------- # DASHBOARD ZE STATYSTYKAMI # ------------------- @app.route('/dashboard') def dashboard(): if 'user_id' not in session: return redirect(url_for('login')) user_id = session['user_id'] user = db.session.get(User, user_id) logs = DeployLog.query.filter_by(user_id=user_id).order_by(DeployLog.timestamp.desc()).all() stats = get_statistics(user_id) for log in logs: log.details = log.details.replace(f" for user {user_id}", "") #pass return render_template('dashboard.html', user=user, logs=logs, stats=stats) # ------------------- # SCHEDULER - AUTOMATYCZNE WDROŻENIA # ------------------- @app.route('/regex-hosts') def list_regex_hosts(): if 'user_id' not in session: return redirect(url_for('login')) user_id = session['user_id'] entries = RegexHostEntry.query.filter_by(user_id=user_id).all() return render_template('list_regex_hosts.html', entries=entries) @app.route('/regex-hosts/new', methods=['GET', 'POST']) def new_regex_host(): if 'user_id' not in session: return redirect(url_for('login')) user_id = session['user_id'] if request.method == 'POST': cidr_range = request.form.get('cidr_range', '').strip() gateway_ip = request.form.get('gateway_ip', '').strip() gateway_hostname = request.form.get('gateway_hostname', '').strip() domain_suffix = request.form.get('domain_suffix', '').strip() host_prefix = request.form.get('host_prefix', '').strip() use_gateway_ip = bool(request.form.get('use_gateway_ip')) comment = request.form.get('comment', '').strip() if not cidr_range: flash('Please provide a CIDR range (e.g. 10.87.200.0/27).', 'danger') return redirect(url_for('new_regex_host')) if not domain_suffix: domain_suffix = "domain.com" if not host_prefix: host_prefix = "ip" entry = RegexHostEntry( user_id=user_id, cidr_range=cidr_range, gateway_ip=gateway_ip, gateway_hostname=gateway_hostname, domain_suffix=domain_suffix, host_prefix=host_prefix, use_gateway_ip=use_gateway_ip, comment=comment ) db.session.add(entry) db.session.commit() flash('New CIDR entry with optional gateway has been added.', 'success') return redirect(url_for('list_regex_hosts')) return render_template('new_edit_regex_host.html', entry=None) @app.route('/regex-hosts//edit', methods=['GET', 'POST']) def edit_regex_host(entry_id): if 'user_id' not in session: return redirect(url_for('login')) entry = db.session.get(RegexHostEntry, entry_id) if not entry or entry.user_id != session['user_id']: flash('cant find row or row is empty', 'danger') return redirect(url_for('list_regex_hosts')) if request.method == 'POST': cidr_range = request.form.get('cidr_range', '').strip() gateway_ip = request.form.get('gateway_ip', '').strip() gateway_hostname = request.form.get('gateway_hostname', '').strip() domain_suffix = request.form.get('domain_suffix', '').strip() host_prefix = request.form.get('host_prefix', '').strip() use_gateway_ip = bool(request.form.get('use_gateway_ip')) comment = request.form.get('comment', '').strip() if not cidr_range: flash('CIDR is required', 'danger') return redirect(url_for('edit_regex_host', entry_id=entry_id)) entry.cidr_range = cidr_range entry.gateway_ip = gateway_ip entry.gateway_hostname = gateway_hostname entry.domain_suffix = domain_suffix or "domain.com" entry.host_prefix = host_prefix or "ip" entry.use_gateway_ip = use_gateway_ip entry.comment = comment db.session.commit() flash('Updated', 'success') return redirect(url_for('list_regex_hosts')) return render_template('new_edit_regex_host.html', entry=entry) @app.route('/regex-hosts//delete', methods=['POST']) def delete_regex_host(entry_id): if 'user_id' not in session: return redirect(url_for('login')) entry = db.session.get(RegexHostEntry, entry_id) if not entry or entry.user_id != session['user_id']: flash('cant find row or row is empty', 'danger') return redirect(url_for('list_regex_hosts')) db.session.delete(entry) db.session.commit() flash('Row deleted', 'info') return redirect(url_for('list_regex_hosts')) @app.route('/delete-selected-backups', methods=['POST']) def delete_selected_backups(): if 'user_id' not in session: return redirect(url_for('login')) selected_ids = request.form.getlist('selected_backups') for backup_id in selected_ids: backup = db.session.get(Backup, backup_id) if backup and backup.user_id == session['user_id']: db.session.delete(backup) db.session.commit() flash('Zaznaczone backupy zostały usunięte.', 'info') return redirect(url_for('backups')) @app.route('/update-host-automation/', methods=['POST']) def update_host_automation(id): if 'user_id' not in session: return redirect(url_for('login')) host = db.session.get(Host, id) if not host or host.user_id != session['user_id']: flash('Serwer nie istnieje lub nie masz uprawnień', 'danger') return redirect(url_for('server_list')) setting = request.form.get('setting') enabled = request.form.get('enabled') == '1' if setting == 'auto_deploy': host.auto_deploy_enabled = enabled elif setting == 'auto_backup': host.auto_backup_enabled = enabled elif setting == 'disable_regex': host.disable_regex_deploy = enabled elif setting == 'disable_local_default': host.disable_local_default = enabled db.session.commit() flash('Ustawienia automatyzacji zostały zaktualizowane.', 'success') return redirect(url_for('server_list')) # ------------------- # EDYCJA LOKALNEGO PLIKU HOSTS # ------------------- @app.route('/edit-local-hosts', methods=['GET', 'POST'], endpoint='edit_local_hosts') def edit_local_hosts(): if 'user_id' not in session: return redirect(url_for('login')) user_id = session['user_id'] hostfile = HostFile.query.filter_by(user_id=user_id, title="Default Hosts").first() if not hostfile: default_content = "# This is a sample hosts file.\n127.0.0.1 localhost\n" hostfile = HostFile(user_id=user_id, title="Default Hosts", content=default_content) db.session.add(hostfile) db.session.commit() if request.method == 'POST': new_content = request.form['hosts_content'] # Zapisz obecną wersję do historii przed zmianą version = HostFileVersion(hostfile_id=hostfile.id, content=hostfile.content) db.session.add(version) # Aktualizacja treści hostfile.content = new_content db.session.commit() flash('Local hosts content updated successfully and previous version saved.', 'success') return render_template('edit_hosts.html', content=hostfile.content, hostfile=hostfile) @app.route('/hostfile//versions', methods=['GET', 'POST']) def hostfile_versions(hostfile_id): if 'user_id' not in session: return redirect(url_for('login')) hostfile = HostFile.query.get(hostfile_id) if not hostfile or hostfile.user_id != session['user_id']: flash('Hostfile not found or unauthorized', 'danger') return redirect(url_for('dashboard')) if request.method == 'POST': selected_ids = request.form.getlist('selected_versions') for version_id in selected_ids: version = HostFileVersion.query.get(version_id) if version and version.hostfile.user_id == session['user_id']: db.session.delete(version) db.session.commit() flash('Wybrane wersje zostały usunięte.', 'info') return redirect(url_for('hostfile_versions', hostfile_id=hostfile_id)) versions = HostFileVersion.query.filter_by(hostfile_id=hostfile.id)\ .order_by(HostFileVersion.timestamp.desc()).all() return render_template('hostfile_versions.html', hostfile=hostfile, versions=versions) @app.route('/hostfile//versions/delete_old/') def delete_old_versions(hostfile_id, days): if 'user_id' not in session: return redirect(url_for('login')) hostfile = HostFile.query.get(hostfile_id) if not hostfile or hostfile.user_id != session['user_id']: flash('Hostfile not found or unauthorized', 'danger') return redirect(url_for('dashboard')) cutoff = datetime.now(timezone.utc) - timedelta(days=days) old_versions = HostFileVersion.query.filter(HostFileVersion.hostfile_id == hostfile_id, HostFileVersion.timestamp < cutoff).all() for version in old_versions: db.session.delete(version) db.session.commit() flash(f'Usunięto wersje starsze niż {days} dni.', 'info') return redirect(url_for('hostfile_versions', hostfile_id=hostfile_id)) @app.route('/hostfile/diff_current/') def diff_current_hostfile(hostfile_id): if 'user_id' not in session: return redirect(url_for('login')) hostfile = HostFile.query.get(hostfile_id) if not hostfile or hostfile.user_id != session['user_id']: flash('Hostfile not found or unauthorized.', 'danger') return redirect(url_for('dashboard')) # Używamy len() zamiast |length if not hostfile.versions or len(hostfile.versions) == 0: flash('Brak zapisanej historii wersji do porównania.', 'warning') return redirect(url_for('hostfile_versions', hostfile_id=hostfile_id)) latest_version = hostfile.versions[0] differ = difflib.HtmlDiff(wrapcolumn=80) diff_html = differ.make_table( latest_version.content.splitlines(), hostfile.content.splitlines(), fromdesc=f"Najnowsza wersja historii (ID: {latest_version.id}, {latest_version.timestamp.strftime('%Y-%m-%d %H:%M:%S')})", todesc="Aktualna zawartość", context=True, numlines=3 ) return render_template('diff_versions.html', diff_html=diff_html, hostfile_id=hostfile.id) @app.route('/hostfile/diff//') def diff_hostfile_versions(version1_id, version2_id): if 'user_id' not in session: return redirect(url_for('login')) version1 = HostFileVersion.query.get(version1_id) version2 = HostFileVersion.query.get(version2_id) if not version1 or not version2 or version1.hostfile.user_id != session['user_id'] or version2.hostfile.user_id != session['user_id']: flash('Wersje nie znalezione lub brak uprawnień.', 'danger') return redirect(url_for('dashboard')) differ = difflib.HtmlDiff(wrapcolumn=80) diff_html = differ.make_table( version1.content.splitlines(), version2.content.splitlines(), fromdesc=f"Wersja {version1.id} - {version1.timestamp.strftime('%Y-%m-%d %H:%M:%S')}", todesc=f"Wersja {version2.id} - {version2.timestamp.strftime('%Y-%m-%d %H:%M:%S')}", context=True, numlines=3 ) hostfile_id = version1.hostfile_id return render_template('diff_versions.html', diff_html=diff_html, hostfile_id=hostfile_id) @app.route('/hostfile/version/') def view_hostfile_version(version_id): if 'user_id' not in session: return redirect(url_for('login')) version = HostFileVersion.query.get(version_id) if not version or version.hostfile.user_id != session['user_id']: flash('Version not found or unauthorized', 'danger') return redirect(url_for('dashboard')) return render_template('view_hostfile_version.html', version=version) @app.route('/hostfile/version//restore') def restore_hostfile_version(version_id): if 'user_id' not in session: return redirect(url_for('login')) version = HostFileVersion.query.get(version_id) if not version or version.hostfile.user_id != session['user_id']: flash('Version not found or unauthorized', 'danger') return redirect(url_for('dashboard')) hostfile = version.hostfile hostfile.content = version.content db.session.commit() flash('Version restored successfully.', 'success') return redirect(url_for('edit_local_hosts')) @app.route('/hostfile/versions') def default_hostfile_versions(): if 'user_id' not in session: return redirect(url_for('login')) # Zakładamy, że domyślny plik hosts ma tytuł "Default Hosts" hostfile = HostFile.query.filter_by(user_id=session['user_id'], title="Default Hosts").first() if not hostfile: flash("Default Hosts file not found.", "danger") return redirect(url_for('edit_local_hosts')) return redirect(url_for('hostfile_versions', hostfile_id=hostfile.id)) def scheduled_deployments(): with app.app_context(): now = datetime.now(timezone.utc) settings_list = UserSettings.query.filter_by(auto_deploy_enabled=True).all() for setting in settings_list: if not setting.deploy_cron: continue if setting.last_deploy_time: base_time = setting.last_deploy_time if base_time.tzinfo is None: base_time = base_time.replace(tzinfo=timezone.utc) else: base_time = datetime(1970,1,1, tzinfo=timezone.utc) cron = croniter(setting.deploy_cron, base_time) next_deploy = cron.get_next(datetime) if now >= next_deploy: deploy_user(setting.user_id) setting.last_deploy_time = now db.session.commit() @app.route('/server-info/', methods=['GET']) def server_info(id): if 'user_id' not in session: return {"error": "Unauthorized"}, 401 host = db.session.get(Host, id) if not host or host.user_id != session['user_id']: return {"error": "Host not found or unauthorized"}, 404 if host.use_daemon and host.type == 'linux': import requests headers = {"Authorization": host.daemon_token} sysinfo_url = host.daemon_url.rstrip('/') + '/system-info' try: resp = requests.get(sysinfo_url, headers=headers, verify=False, timeout=5) if resp.status_code == 200: data = resp.json() return { "hostname": host.resolved_daemon if host.daemon_url else host.resolved_hostname, "ip": host.raw_ip, "cpu": data.get('cpu_percent'), "mem": data.get('memory_percent'), "disk": data.get('disk_percent'), "uptime_seconds": data.get('uptime_seconds') } else: return {"error": f"Błąd demona: {resp.status_code}"}, 500 except Exception as e: return {"error": str(e)}, 500 else: return {"error": "This server do not use daemon."}, 400 @app.errorhandler(404) def page_not_found(error): return render_template("404.html", error=error), 404 @app.errorhandler(500) def internal_server_error(error): if app.debug: return render_template("500.html", error=error), 500 @app.route('/local-defaults', methods=['GET', 'POST']) def local_defaults(): if 'user_id' not in session: return redirect(url_for('login')) user_id = session['user_id'] if request.method == 'POST': hostname = request.form.get('hostname', '').strip() ip_address = request.form.get('ip_address', '').strip() if hostname and ip_address: entry_content = f"{ip_address} {hostname}" new_entry = LocalDefaultEntry(user_id=user_id, hostname=hostname, ip_address=ip_address, entry=entry_content) db.session.add(new_entry) db.session.commit() flash('Dodano nowy wpis.', 'success') else: flash('Hostname i adres IP są wymagane.', 'danger') return redirect(url_for('local_defaults')) entries = LocalDefaultEntry.query.filter_by(user_id=user_id).all() return render_template('local_defaults.html', entries=entries) @app.route('/local-defaults/delete/', methods=['POST']) @app.route('/local-defaults/delete', methods=['POST']) def delete_local_default(entry_id=None): if 'user_id' not in session: return jsonify({'status': 'error', 'message': 'Unauthorized'}), 403 if request.is_json: data = request.get_json() entry_ids = data.get('entry_ids', []) if not entry_ids: return jsonify({'status': 'error', 'message': 'Brak wpisów do usunięcia'}), 400 entries_to_delete = LocalDefaultEntry.query.filter( LocalDefaultEntry.id.in_(entry_ids), LocalDefaultEntry.user_id == session['user_id'] ).all() if not entries_to_delete: return jsonify({'status': 'error', 'message': 'Nie znaleziono wpisów do usunięcia'}), 404 for entry in entries_to_delete: db.session.delete(entry) db.session.commit() return jsonify({'status': 'success', 'message': 'Wpisy usunięte pomyślnie'}) else: # Obsługa usuwania pojedynczego wpisu entry = LocalDefaultEntry.query.get(entry_id) if not entry or entry.user_id != session['user_id']: flash('Wpis nie istnieje lub brak uprawnień.', 'danger') return redirect(url_for('local_defaults')) db.session.delete(entry) db.session.commit() flash('Wpis został usunięty.', 'info') return redirect(url_for('local_defaults')) @app.route('/local-defaults/update/', methods=['POST']) def update_local_default(entry_id): if 'user_id' not in session: return jsonify({'status': 'error', 'message': 'Unauthorized'}), 403 entry = db.session.get(LocalDefaultEntry, entry_id) if not entry or entry.user_id != session['user_id']: return jsonify({'status': 'error', 'message': 'Entry not found'}), 404 data = request.get_json() new_ip = data.get('ip_address', '').strip() new_hostname = data.get('hostname', '').strip() if new_ip and new_hostname: entry.ip_address = new_ip entry.hostname = new_hostname entry.entry = f"{new_ip} {new_hostname}" db.session.commit() return jsonify({'status': 'success', 'message': 'Updated successfully', 'ip_address': new_ip, 'hostname': new_hostname}) else: return jsonify({'status': 'error', 'message': 'IP and hostname are required'}), 400 scheduler = BackgroundScheduler(timezone=get_localzone()) scheduler.add_job(func=scheduled_deployments, trigger="interval", minutes=1, next_run_time=datetime.now()) scheduler.add_job(func=automated_backups, trigger="interval", minutes=1, next_run_time=datetime.now()) scheduler.add_job(func=cleanup_old_backups, trigger="interval", hours=24, next_run_time=datetime.now()) if __name__ == '__main__': with app.app_context(): db.create_all() app.run( host='0.0.0.0', port=5580, use_reloader=False, )