from flask import Flask, render_template, request, redirect, url_for, flash, session, Response 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]) # Funkcja wspolpracy z hosts_daemon use_daemon = db.Column(db.Boolean, default=False) # <--- NOWE daemon_url = db.Column(db.String(255), nullable=True) # <--- NOWE daemon_token = db.Column(db.String(255), nullable=True) # <--- NOWE @property def resolved_hostname(self): try: return socket.gethostbyaddr(self.hostname)[0] except Exception: 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) 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)) # Funkcje pomocnicze def ensure_local_defaults(content): required_lines = [ "127.0.0.1 localhost", "::1 localhost ip6-localhost ip6-loopback", "255.255.255.255 broadcasthost" ] lines = [l.rstrip() for l in content.splitlines()] lines = [line for line in lines if line not in required_lines] lines = required_lines + lines return "\n".join(lines) + "\n" 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 == 'ssh_key' and host_obj.private_key: key_file_obj = io.StringIO(host_obj.private_key) passphrase = host_obj.key_passphrase if host_obj.key_passphrase else None try: pkey = paramiko.RSAKey.from_private_key(key_file_obj, password=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, # TCP connection timeout banner_timeout=30 # Wait longer for SSH banner ) 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 # pobieramy /etc/hosts z demona: url = host.daemon_url.rstrip('/') + '/hosts' # Zmiana: jeśli demon wymaga nagłówka Bearer: 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", "") else: # standard: 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() 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 = Backup( user_id=host.user_id, host_id=host.id, content=content, description=f'Automated backup from {host.hostname} at {datetime.now(timezone.utc).isoformat()}' ) db.session.add(backup) db.session.commit() log_entry = DeployLog(details=f'[BACKUP] Automatic backup created for host {host.hostname}', user_id=host.user_id) db.session.add(log_entry) db.session.commit() print(f'Automated backup for host {host.hostname} created successfully.') except Exception as e: print(f'Error creating automated backup for host {host.hostname}: {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() # ------------------- # 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 HOSTAMI # ------------------- @app.route('/add_server', methods=['GET', 'POST']) def add_server(): if 'user_id' not in session: return redirect(url_for('login')) # Pobieramy wszystkie HostFile tego użytkownika, np. do wyświetlenia w : 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') new_private_key = request.form.get('private_key', '').strip() new_passphrase = request.form.get('key_passphrase', '').strip() if host.auth_method == 'ssh_key' and new_private_key: host.private_key = new_private_key if host.auth_method == 'ssh_key' and new_passphrase: host.key_passphrase = new_passphrase # Demon: 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 # Nowe pole: preferred_hostfile_id 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')) # GET -> renderuj z user_hostfiles 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': # Połączenie przez demon (self-signed certy, verify=False) import requests headers = {"Authorization": host.daemon_token} # Najpierw sprawdzenie /health 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 {host.hostname}', 'success') else: raise Exception(f"Demon health check returned {resp.status_code}") # Dodatkowe pobranie /system-info 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() # Wyświetlamy kilka przykładowych danych w flash: msg = (f"System-info for {host.hostname}: " 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: # Standardowe sprawdzenie przez SSH ssh = open_ssh_connection(host) ssh.close() flash(f'SSH connection to {host.hostname} successful.', 'success') except Exception as e: flash(f'Connection failed for {host.hostname}: {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) @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_servers')) default_content = ensure_local_defaults("") try: if host.use_daemon and host.type == 'linux': import requests url = host.daemon_url.rstrip('/') + '/hosts' headers = {"Authorization": host.daemon_token} # Zakładamy, że demon potrafi zastąpić /etc/hosts treścią "default_content" 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: # standard linux clear_linux(host, default_content) 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('clear_all_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("") for h in hosts: try: 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: {h.hostname}', 'success') elif h.type == 'mikrotik' and mikrotik_clear: clear_mikrotik(h) flash(f'Cleared Mikrotik host: {h.hostname}', 'success') except Exception as e: flash(f'Error clearing host {h.hostname}: {str(e)}', 'danger') return redirect(url_for('clear_all_server')) return render_template('clear_servers.html', hosts=hosts) # ------------------- # 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() if request.method == 'POST': selected_host_ids = request.form.getlist('hosts') for host in hosts: if str(host.id) in selected_host_ids: try: # Przygotuj zawartość do wgrania 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}") db.session.add(DeployLog( details=f'[LINUX/DAEMON] Deployed file "{file.title}" to {host.hostname} for user {session["user_id"]}', user_id=session['user_id'] )) elif host.type == 'mikrotik': # Mikrotik wrapped_mikro = wrap_mikrotik_content(file.content) deploy_mikrotik(host, wrapped_mikro) db.session.add(DeployLog( details=f'[MIKROTIK] Deployed file "{file.title}" to {host.hostname} for user {session["user_id"]}', user_id=session['user_id'] )) else: # Standard Linux (SSH) 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) db.session.add(DeployLog( details=f'[LINUX] Deployed file "{file.title}" to {host.hostname} for user {session["user_id"]}', user_id=session['user_id'] )) db.session.commit() flash(f'Deployed file "{file.title}" to {host.hostname}', 'success') except Exception as e: flash(f'Error deploying file "{file.title}" to {host.hostname}: {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 (daemon) from {host.hostname}' 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 (mikrotik) from {host.hostname}' else: # Standard Linux (SSH) 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 {host.hostname}' 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 {host.hostname} created successfully.', 'success') except Exception as e: flash(f'Error creating backup for host {host.hostname}: {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': ssh = open_ssh_connection(host) # Usuń istniejące wpisy ssh.exec_command("/ip dns static remove [find]") import time time.sleep(1) # Przygotuj jedno polecenie, które dodaje wszystkie wpisy commands = [] for line in backup.content.splitlines(): line = line.strip() if line.startswith("add "): commands.append("/ip dns static " + line) full_command = " ; ".join(commands) #print("[DEBUG] Full command sent to Mikrotik:", full_command) ssh.exec_command(full_command) ssh.close() flash(f'Backup restored to mikrotik host {host.hostname} 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 #print(f"[DEBUG] Tymczasowy plik: {tmp_file_path} zawiera: {backup.content}") sftp.put(tmp_file_path, '/etc/hosts') sftp.close() ssh.close() os.remove(tmp_file_path) flash(f'Backup restored to host {host.hostname} successfully.', 'success') except Exception as e: flash(f'Error restoring backup to host {host.hostname}: {str(e)}', 'danger') else: # Przywrócenie backupu jako domyślnej konfiguracji (Default Hosts) 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 (daemon) from {host.hostname}' 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 (mikrotik) from {host.hostname}' else: # Standard Linux (SSH) 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 {host.hostname}' 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 host {host.hostname}: {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) cw.writerow(['hostname', 'username', 'password', 'port', 'type', 'auth_method', 'private_key', 'key_passphrase']) for host in hosts: cw.writerow([host.hostname, host.username, host.password, host.port, host.type, host.auth_method, host.private_key or '', host.key_passphrase or '']) output = si.getvalue() return Response(output, mimetype="text/csv", headers={"Content-Disposition": "attachment;filename=servers.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) for row in csv_input: if len(row) < 8: continue hostname, username, password_val, port_str, host_type, auth_method, private_key, key_passphrase = row try: port = int(port_str) except ValueError: port = 22 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, 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') # 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 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')) # ------------------- # 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'] # hostfile.content = new_content # db.session.commit() # flash('Local hosts content updated successfully', 'success') # return render_template('edit_hosts.html', content=hostfile.content) # ------------------- # DEPLOYMENT DOMYŚLNY DLA UŻYTKOWNIKA # ------------------- def deploy_user(user_id): user_settings = UserSettings.query.filter_by(user_id=user_id).first() # domyślny plik "Default Hosts" (np. szukasz title=="Default Hosts") default_file = HostFile.query.filter_by(user_id=user_id, title="Default Hosts").first() if not default_file: # jeśli nie ma w ogóle, nic nie robimy return # ewentualnie regex 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 # Który plik wgrywać? if h.preferred_hostfile_id: chosen_file = HostFile.query.filter_by(id=h.preferred_hostfile_id, user_id=user_id).first() if not chosen_file: # fallback do default chosen_file = default_file else: chosen_file = default_file # final_content final_content = regex_lines + chosen_file.content try: if h.type == 'mikrotik': wrapped_content = wrap_mikrotik_content(final_content) deploy_mikrotik(h, wrapped_content) db.session.add(DeployLog( details=f'[MIKROTIK] Updated {h.hostname} for user {user_id}', user_id=user_id )) elif h.use_daemon and h.type == 'linux': # Demon import requests adjusted_content = ensure_local_defaults(final_content) wrapped_content = wrap_content_with_comments(adjusted_content) url = h.daemon_url.rstrip('/') + '/hosts' headers = {"Authorization": h.daemon_token} resp = requests.post(url, json={"hosts": wrapped_content}, headers=headers, timeout=10) if resp.status_code != 200: raise Exception(f"Daemon POST error: {resp.status_code} - {resp.text}") db.session.add(DeployLog( details=f'[LINUX/DAEMON] Updated {h.hostname} for user {user_id}', user_id=user_id )) else: # standard linux - SSH ssh = open_ssh_connection(h) adjusted_content = ensure_local_defaults(final_content) wrapped_content = wrap_content_with_comments(adjusted_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) db.session.add(DeployLog( details=f'[LINUX] Updated {h.hostname} for user {user_id}', user_id=user_id )) db.session.commit() except Exception as e: db.session.add(DeployLog( details=f'Failed to update {h.hostname}: {str(e)} for user {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 = {} for line in exported: line = line.strip() if not line.startswith('add '): continue 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: existing_dns[name_val] = address_val desired_dns = {} for line in hosts_content.splitlines(): line = line.strip() if (not line or line.startswith('#') or 'Auto-hosts_upload:' in line or 'End_of_auto-hosts_upload:' in line): continue parts = line.split() if len(parts) < 2: continue ip_address = parts[0] hostnames = parts[1:] for hname in hostnames: desired_dns[hname] = ip_address 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, existing_ip in existing_dns.items(): 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}", "") 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 db.session.commit() flash('Ustawienia automatyzacji zostały zaktualizowane.', 'success') return redirect(url_for('server_list')) @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': # Masowe usuwanie – lista zaznaczonych wersji 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 ) # Przekazujemy hostfile_id, zakładając, że obie wersje należą do tego samego pliku 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')) # Przywróć zawartość wersji do głównego hostfile 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() 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, )