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/<int:id>')
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/<int:id>', 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/<int:id>', 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/<int:host_id>', 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/<int:file_id>/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/<int:file_id>/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/<int:file_id>', 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/<int:host_id>', 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/<int:backup_id>', 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)
ssh.exec_command("/ip dns static remove [find]")
import time
time.sleep(1)
commands = []
for line in backup.content.splitlines():
line = line.strip()
if line.startswith("add "):
commands.append("/ip dns static " + line)
full_command = " ; ".join(commands)
ssh.exec_command(full_command)
ssh.close()
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/<int:backup_id>', 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/<int:id>', 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/<int:backup_id>', 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 = {}
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}", "")
#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/<int:entry_id>/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/<int:entry_id>/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/<int:id>', 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/<int:hostfile_id>/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/<int:hostfile_id>/versions/delete_old/<int:days>')
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/<int:hostfile_id>')
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/<int:version1_id>/<int:version2_id>')
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/<int:version_id>')
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/<int:version_id>/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/<int:id>', 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/<int:entry_id>', 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/<int:entry_id>', 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,
)