454 lines
15 KiB
Python
454 lines
15 KiB
Python
from flask import Blueprint, render_template, redirect, url_for, request, flash, jsonify
|
|
from flask_login import login_required, current_user
|
|
import markdown
|
|
from app.core.models import Server, App, Subnet, Port
|
|
from app.core.extensions import db, limiter
|
|
from datetime import datetime
|
|
from app.utils.app_utils import validate_app_data, is_port_in_use
|
|
|
|
bp = Blueprint("dashboard", __name__, url_prefix="/dashboard")
|
|
|
|
|
|
@bp.route("/")
|
|
def dashboard_home():
|
|
"""Main dashboard view showing server statistics"""
|
|
# Check if user is logged in, redirect if not
|
|
if not current_user.is_authenticated:
|
|
flash("Please log in to access this page.", "info")
|
|
return redirect(url_for('auth.login'))
|
|
|
|
server_count = Server.query.count()
|
|
app_count = App.query.count()
|
|
subnet_count = Subnet.query.count()
|
|
|
|
# Get latest added servers
|
|
latest_servers = Server.query.order_by(Server.created_at.desc()).limit(5).all()
|
|
|
|
# Get subnets with usage stats
|
|
subnets = Subnet.query.all()
|
|
for subnet in subnets:
|
|
subnet.usage_percent = (
|
|
subnet.used_ips / 254 * 100 if subnet.cidr.endswith("/24") else 0
|
|
)
|
|
|
|
return render_template(
|
|
"dashboard/index.html",
|
|
title="Dashboard",
|
|
server_count=server_count,
|
|
app_count=app_count,
|
|
subnet_count=subnet_count,
|
|
latest_servers=latest_servers,
|
|
subnets=subnets,
|
|
now=datetime.now(),
|
|
)
|
|
|
|
|
|
@bp.route("/servers")
|
|
@login_required
|
|
def server_list():
|
|
"""List all servers"""
|
|
servers = Server.query.order_by(Server.hostname).all()
|
|
|
|
return render_template(
|
|
"dashboard/server_list.html",
|
|
title="Servers",
|
|
servers=servers,
|
|
now=datetime.now(),
|
|
)
|
|
|
|
|
|
@bp.route("/server/<int:server_id>")
|
|
@login_required
|
|
def server_view(server_id):
|
|
"""View server details"""
|
|
server = Server.query.get_or_404(server_id)
|
|
apps = App.query.filter_by(server_id=server_id).all()
|
|
|
|
return render_template(
|
|
"dashboard/server_view.html",
|
|
title=f"Server - {server.hostname}",
|
|
server=server,
|
|
apps=apps,
|
|
now=datetime.now(),
|
|
)
|
|
|
|
|
|
@bp.route("/server/new", methods=["GET", "POST"])
|
|
@login_required
|
|
def server_new():
|
|
"""Create a new server"""
|
|
subnets = Subnet.query.all()
|
|
|
|
if request.method == "POST":
|
|
hostname = request.form.get("hostname")
|
|
ip_address = request.form.get("ip_address")
|
|
subnet_id = request.form.get("subnet_id")
|
|
documentation = request.form.get("documentation", "")
|
|
|
|
# Basic validation
|
|
if not hostname or not ip_address or not subnet_id:
|
|
flash("Please fill in all required fields", "danger")
|
|
return render_template(
|
|
"dashboard/server_form.html",
|
|
title="New Server",
|
|
subnets=subnets,
|
|
now=datetime.now(),
|
|
)
|
|
|
|
# Check if hostname or IP already exists
|
|
if Server.query.filter_by(hostname=hostname).first():
|
|
flash("Hostname already exists", "danger")
|
|
return render_template(
|
|
"dashboard/server_form.html",
|
|
title="New Server",
|
|
subnets=subnets,
|
|
now=datetime.now(),
|
|
)
|
|
|
|
if Server.query.filter_by(ip_address=ip_address).first():
|
|
flash("IP address already exists", "danger")
|
|
return render_template(
|
|
"dashboard/server_form.html",
|
|
title="New Server",
|
|
subnets=subnets,
|
|
now=datetime.now(),
|
|
)
|
|
|
|
# Create new server
|
|
server = Server(
|
|
hostname=hostname,
|
|
ip_address=ip_address,
|
|
subnet_id=subnet_id,
|
|
documentation=documentation,
|
|
)
|
|
|
|
db.session.add(server)
|
|
db.session.commit()
|
|
|
|
flash("Server created successfully", "success")
|
|
return redirect(url_for("dashboard.server_view", server_id=server.id))
|
|
|
|
return render_template(
|
|
"dashboard/server_form.html",
|
|
title="New Server",
|
|
subnets=subnets,
|
|
now=datetime.now(),
|
|
)
|
|
|
|
|
|
@bp.route("/server/<int:server_id>/edit", methods=["GET", "POST"])
|
|
@login_required
|
|
def server_edit(server_id):
|
|
"""Edit an existing server"""
|
|
server = Server.query.get_or_404(server_id)
|
|
subnets = Subnet.query.all()
|
|
|
|
if request.method == "POST":
|
|
hostname = request.form.get("hostname")
|
|
ip_address = request.form.get("ip_address")
|
|
subnet_id = request.form.get("subnet_id")
|
|
documentation = request.form.get("documentation", "")
|
|
|
|
if not hostname or not ip_address or not subnet_id:
|
|
flash("All fields are required", "danger")
|
|
return render_template(
|
|
"dashboard/server_form.html",
|
|
title="Edit Server",
|
|
server=server,
|
|
subnets=subnets,
|
|
)
|
|
|
|
# Check if hostname changed and already exists
|
|
if (
|
|
hostname != server.hostname
|
|
and Server.query.filter_by(hostname=hostname).first()
|
|
):
|
|
flash("Hostname already exists", "danger")
|
|
return render_template(
|
|
"dashboard/server_form.html",
|
|
title="Edit Server",
|
|
server=server,
|
|
subnets=subnets,
|
|
)
|
|
|
|
# Check if IP changed and already exists
|
|
if (
|
|
ip_address != server.ip_address
|
|
and Server.query.filter_by(ip_address=ip_address).first()
|
|
):
|
|
flash("IP address already exists", "danger")
|
|
return render_template(
|
|
"dashboard/server_form.html",
|
|
title="Edit Server",
|
|
server=server,
|
|
subnets=subnets,
|
|
)
|
|
|
|
# Update server
|
|
server.hostname = hostname
|
|
server.ip_address = ip_address
|
|
server.subnet_id = subnet_id
|
|
server.documentation = documentation
|
|
|
|
db.session.commit()
|
|
|
|
flash("Server updated successfully", "success")
|
|
return redirect(url_for("dashboard.server_view", server_id=server.id))
|
|
|
|
# GET request - show form with current values
|
|
return render_template(
|
|
"dashboard/server_form.html",
|
|
title=f"Edit Server - {server.hostname}",
|
|
server=server,
|
|
subnets=subnets,
|
|
)
|
|
|
|
|
|
@bp.route("/server/<int:server_id>/delete", methods=["POST"])
|
|
@login_required
|
|
def server_delete(server_id):
|
|
"""Delete a server"""
|
|
server = Server.query.get_or_404(server_id)
|
|
|
|
# Delete all apps associated with this server
|
|
App.query.filter_by(server_id=server_id).delete()
|
|
|
|
# Delete the server
|
|
db.session.delete(server)
|
|
db.session.commit()
|
|
|
|
flash("Server deleted successfully", "success")
|
|
return redirect(url_for("dashboard.dashboard_home"))
|
|
|
|
|
|
@bp.route("/app/new", methods=["GET", "POST"])
|
|
@bp.route("/app/new/<int:server_id>", methods=["GET", "POST"])
|
|
@login_required
|
|
def app_new(server_id=None):
|
|
"""Create a new application"""
|
|
servers = Server.query.all()
|
|
|
|
if request.method == "POST":
|
|
name = request.form.get("name", "").strip()
|
|
server_id = request.form.get("server_id")
|
|
documentation = request.form.get("documentation", "")
|
|
url = request.form.get("url", "")
|
|
|
|
# Process port data from form
|
|
port_data = []
|
|
port_numbers = request.form.getlist("port_numbers[]")
|
|
protocols = request.form.getlist("protocols[]")
|
|
descriptions = request.form.getlist("port_descriptions[]")
|
|
|
|
for i in range(len(port_numbers)):
|
|
if port_numbers[i] and port_numbers[i].strip():
|
|
protocol = protocols[i] if i < len(protocols) else "TCP"
|
|
description = descriptions[i] if i < len(descriptions) else ""
|
|
port_data.append((port_numbers[i], protocol, description))
|
|
|
|
# Server-side validation
|
|
from app.utils.app_utils import save_app
|
|
|
|
success, app, error = save_app(name, server_id, documentation, port_data, url=url)
|
|
|
|
if success:
|
|
flash(f"Application '{name}' created successfully", "success")
|
|
return redirect(url_for("dashboard.app_view", app_id=app.id))
|
|
else:
|
|
flash(error, "danger")
|
|
|
|
# Check if it's a port conflict and provide direct link
|
|
if "already in use by application" in error:
|
|
try:
|
|
app_name = error.split("'")[1] # Extract app name from error
|
|
conflict_app = App.query.filter_by(name=app_name, server_id=server_id).first()
|
|
if conflict_app:
|
|
edit_url = url_for('dashboard.app_edit', app_id=conflict_app.id)
|
|
flash(f'<a href="{edit_url}" class="alert-link">Edit conflicting application</a>', "info")
|
|
except:
|
|
pass
|
|
|
|
# GET request or validation failed - render the form
|
|
return render_template(
|
|
"dashboard/app_form.html",
|
|
title="New Application",
|
|
edit_mode=False,
|
|
servers=servers,
|
|
selected_server_id=server_id,
|
|
)
|
|
|
|
|
|
@bp.route("/app/<int:app_id>", methods=["GET"])
|
|
@login_required
|
|
def app_view(app_id):
|
|
"""View a specific application"""
|
|
app = App.query.get_or_404(app_id)
|
|
server = Server.query.get(app.server_id)
|
|
|
|
return render_template(
|
|
"dashboard/app_view.html",
|
|
title=f"Application - {app.name}",
|
|
app=app,
|
|
server=server,
|
|
now=datetime.now(),
|
|
)
|
|
|
|
|
|
@bp.route("/app/<int:app_id>/edit", methods=["GET", "POST"])
|
|
@login_required
|
|
def app_edit(app_id):
|
|
"""Edit an existing application with comprehensive error handling"""
|
|
# Get the application and all servers
|
|
app = App.query.get_or_404(app_id)
|
|
servers = Server.query.all()
|
|
|
|
if request.method == "POST":
|
|
# Get form data
|
|
name = request.form.get("name", "").strip()
|
|
server_id = request.form.get("server_id")
|
|
documentation = request.form.get("documentation", "")
|
|
url = request.form.get("url", "") # Get the URL
|
|
|
|
# Process port data from form
|
|
port_data = []
|
|
port_numbers = request.form.getlist("port_numbers[]")
|
|
protocols = request.form.getlist("protocols[]")
|
|
descriptions = request.form.getlist("port_descriptions[]")
|
|
|
|
for i in range(len(port_numbers)):
|
|
if port_numbers[i] and port_numbers[i].strip():
|
|
protocol = protocols[i] if i < len(protocols) else "TCP"
|
|
description = descriptions[i] if i < len(descriptions) else ""
|
|
port_data.append((port_numbers[i], protocol, description))
|
|
|
|
# Check for port conflicts proactively
|
|
conflicts = []
|
|
for i, (port_number, protocol, _) in enumerate(port_data):
|
|
try:
|
|
clean_port = int(port_number)
|
|
in_use, conflicting_app_name = is_port_in_use(
|
|
clean_port, protocol, server_id, exclude_app_id=app_id
|
|
)
|
|
|
|
if in_use:
|
|
conflicts.append((clean_port, protocol, conflicting_app_name))
|
|
except (ValueError, TypeError):
|
|
continue
|
|
|
|
if conflicts:
|
|
# Find the IDs of conflicting apps for linking
|
|
conflict_msgs = []
|
|
for port, protocol, conflict_app_name in conflicts:
|
|
conflict_app = App.query.filter_by(name=conflict_app_name, server_id=server_id).first()
|
|
if conflict_app:
|
|
edit_url = url_for('dashboard.app_edit', app_id=conflict_app.id)
|
|
conflict_msgs.append(
|
|
f'Port {port}/{protocol} is in use by <a href="{edit_url}">{conflict_app_name}</a>'
|
|
)
|
|
else:
|
|
conflict_msgs.append(f'Port {port}/{protocol} is in use by {conflict_app_name}')
|
|
|
|
for msg in conflict_msgs:
|
|
flash(msg, "danger")
|
|
|
|
return render_template(
|
|
"dashboard/app_form.html",
|
|
title=f"Edit {app.name}",
|
|
edit_mode=True,
|
|
servers=servers,
|
|
app=app
|
|
)
|
|
|
|
# Replace local validation with shared function
|
|
valid, error = validate_app_data(name, server_id, existing_app_id=app_id)
|
|
|
|
if valid:
|
|
# Update application with URL
|
|
app.name = name
|
|
app.server_id = server_id
|
|
app.documentation = documentation
|
|
app.url = url
|
|
|
|
# Update application
|
|
from app.utils.app_utils import save_app
|
|
|
|
success, updated_app, error = save_app(
|
|
name, server_id, documentation, port_data, app_id, url
|
|
)
|
|
|
|
if success:
|
|
flash("Application updated successfully", "success")
|
|
return redirect(url_for("dashboard.app_view", app_id=app_id))
|
|
else:
|
|
flash(error, "danger")
|
|
|
|
# Extract app name from error and provide link if it's a conflict
|
|
if "already in use by application" in error:
|
|
app_name = error.split("'")[1] # Extract app name from error message
|
|
conflict_app = App.query.filter_by(name=app_name, server_id=server_id).first()
|
|
if conflict_app:
|
|
edit_url = url_for('dashboard.app_edit', app_id=conflict_app.id)
|
|
flash(
|
|
f'Would you like to edit the conflicting application? '
|
|
f'<a href="{edit_url}">Edit {app_name}</a>',
|
|
"info"
|
|
)
|
|
else:
|
|
flash(error, "danger")
|
|
|
|
# GET request - display the form
|
|
return render_template(
|
|
"dashboard/app_form.html",
|
|
title=f"Edit {app.name}",
|
|
edit_mode=True,
|
|
servers=servers,
|
|
app=app
|
|
)
|
|
|
|
|
|
@bp.route("/app/<int:app_id>/delete", methods=["POST"])
|
|
@login_required
|
|
def app_delete(app_id):
|
|
"""Delete an application"""
|
|
app = App.query.get_or_404(app_id)
|
|
server_id = app.server_id
|
|
|
|
db.session.delete(app)
|
|
db.session.commit()
|
|
|
|
flash("Application deleted successfully", "success")
|
|
return redirect(url_for("dashboard.server_view", server_id=server_id))
|
|
|
|
|
|
@bp.route("/settings", methods=["GET", "POST"])
|
|
@login_required
|
|
def settings():
|
|
"""User settings page"""
|
|
if request.method == "POST":
|
|
# Handle user settings update
|
|
current_password = request.form.get("current_password")
|
|
new_password = request.form.get("new_password")
|
|
confirm_password = request.form.get("confirm_password")
|
|
|
|
# Validate inputs
|
|
if not current_password:
|
|
flash("Current password is required", "danger")
|
|
return redirect(url_for("dashboard.settings"))
|
|
|
|
if new_password != confirm_password:
|
|
flash("New passwords do not match", "danger")
|
|
return redirect(url_for("dashboard.settings"))
|
|
|
|
# Verify current password
|
|
if not current_user.check_password(current_password):
|
|
flash("Current password is incorrect", "danger")
|
|
return redirect(url_for("dashboard.settings"))
|
|
|
|
# Update password
|
|
current_user.set_password(new_password)
|
|
db.session.commit()
|
|
|
|
flash("Password updated successfully", "success")
|
|
return redirect(url_for("dashboard.settings"))
|
|
|
|
return render_template("dashboard/settings.html", title="User Settings")
|