websites/homedocs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

wip

Browse source
This commit is contained in:
pika 2025-03-31 13:47:34 +02:00
parent 5a309a0f6d
commit 1ab129b798
5 changed files with 499 additions and 107 deletions
Download patch file Download diff file Expand all files Collapse all files

102
app/routes/api.py
View file

@ -295,25 +295,36 @@ def add_app_port(app_id):
protocol = request.form.get("protocol", "TCP")
description = request.form.get("description", "")
# Validate port data
# Validate port data with server-side conflict check
valid, clean_port, error = validate_port_data(
port_number, protocol, description
port_number, protocol, description, app.server_id, app_id
)
if not valid:
flash(error, "danger")
# If port is in use by another app, provide a direct link
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=app.server_id).first()
if conflict_app:
edit_url = url_for('dashboard.app_edit', app_id=conflict_app.id)
edit_link = f'<a href="{edit_url}">Edit {app_name}</a>'
flash(f"Would you like to edit the conflicting application? {edit_link}", "info")
return (
redirect(url_for("dashboard.app_view", app_id=app_id))
if not is_ajax
else jsonify({"success": False, "error": error})
), 400
# Check if port already exists
# Check if port already exists for this app
existing_port = Port.query.filter_by(
app_id=app_id, port_number=clean_port
app_id=app_id, port_number=clean_port, protocol=protocol
).first()
if existing_port:
error_msg = f"Port {clean_port} already exists for this application"
error_msg = f"Port {clean_port}/{protocol} already exists for this application"
flash(error_msg, "warning")
return (
redirect(url_for("dashboard.app_view", app_id=app_id))
@ -457,3 +468,84 @@ def get_free_port(server_id):
return jsonify(
{"success": False, "error": "No free ports available in the range 8000-9000"}
)
@bp.route("/validate/app-name", methods=["GET"])
@login_required
def validate_app_name():
"""API endpoint to validate application name in real-time"""
name = request.args.get("name", "").strip()
server_id = request.args.get("server_id")
app_id = request.args.get("app_id")
if not name or not server_id:
return jsonify({"valid": False, "message": "Missing required parameters"})
# Check for existing app with same name
query = App.query.filter(App.name == name, App.server_id == server_id)
if app_id:
query = query.filter(App.id != int(app_id))
existing_app = query.first()
if existing_app:
return jsonify({
"valid": False,
"message": f"Application '{name}' already exists on this server",
"app_id": existing_app.id,
"edit_url": url_for('dashboard.app_edit', app_id=existing_app.id)
})
# Check for similar names
similar_apps = App.query.filter(
App.name.ilike(f"%{name}%"),
App.server_id == server_id
).limit(3).all()
if similar_apps and (not app_id or not any(str(app.id) == app_id for app in similar_apps)):
similar_names = [{"name": app.name, "id": app.id} for app in similar_apps]
return jsonify({
"valid": True,
"warning": "Similar application names found",
"similar_apps": similar_names
})
return jsonify({"valid": True})
@bp.route("/validate/app-port", methods=["GET"])
@login_required
def validate_app_port():
"""API endpoint to validate port in real-time"""
port_number = request.args.get("port_number", "").strip()
protocol = request.args.get("protocol", "TCP")
server_id = request.args.get("server_id")
app_id = request.args.get("app_id")
if not port_number or not server_id:
return jsonify({"valid": False, "message": "Missing required parameters"})
try:
clean_port = int(port_number)
if clean_port < 1 or clean_port > 65535:
return jsonify({"valid": False, "message": "Port must be between 1 and 65535"})
in_use, app_name = is_port_in_use(
clean_port, protocol, server_id,
exclude_app_id=app_id if app_id else None
)
if in_use:
conflict_app = App.query.filter_by(name=app_name, server_id=server_id).first()
return jsonify({
"valid": False,
"message": f"Port {clean_port}/{protocol} is already in use by application '{app_name}'",
"app_id": conflict_app.id if conflict_app else None,
"edit_url": url_for('dashboard.app_edit', app_id=conflict_app.id) if conflict_app else None
})
return jsonify({"valid": True})
except ValueError:
return jsonify({"valid": False, "message": "Invalid port number"})

144
app/routes/dashboard.py
View file

@ -4,7 +4,7 @@ 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
from app.utils.app_utils import validate_app_data, is_port_in_use
bp = Blueprint("dashboard", __name__, url_prefix="/dashboard")
@ -223,61 +223,54 @@ def server_delete(server_id):
def app_new(server_id=None):
"""Create a new application"""
servers = Server.query.all()
if request.method == "POST":
# Handle form submission
name = request.form.get("name")
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
if not name or not server_id:
flash("Name and server are required", "danger")
return render_template(
"dashboard/app_form.html",
title="New Application",
edit_mode=False,
servers=servers,
selected_server_id=server_id,
)
# Create the app with the URL
app = App(name=name, server_id=server_id, documentation=documentation, url=url)
try:
db.session.add(app)
db.session.commit()
# Process port numbers if any
port_numbers = request.form.getlist("port_numbers[]")
protocols = request.form.getlist("protocols[]")
descriptions = request.form.getlist("port_descriptions[]")
for i, port_number in enumerate(port_numbers):
if port_number and port_number.isdigit():
port = Port(
app_id=app.id,
port_number=int(port_number),
protocol=protocols[i] if i < len(protocols) else "TCP",
description=descriptions[i] if i < len(descriptions) else "",
)
db.session.add(port)
db.session.commit()
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))
except Exception as e:
db.session.rollback()
flash(f"Error creating application: {str(e)}", "danger")
# GET request - render the form
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, # This will pre-select the server
selected_server_id=server_id,
)
@ -324,6 +317,44 @@ def app_edit(app_id):
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)
@ -338,7 +369,7 @@ def app_edit(app_id):
from app.utils.app_utils import save_app
success, updated_app, error = save_app(
name, server_id, documentation, port_data, app_id
name, server_id, documentation, port_data, app_id, url
)
if success:
@ -346,17 +377,28 @@ def app_edit(app_id):
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")
# For GET requests or failed POSTs
# GET request - display the form
return render_template(
"dashboard/app_form.html",
title=f"Edit Application: {app.name}",
title=f"Edit {app.name}",
edit_mode=True,
app=app,
dashboard_link=url_for("dashboard.dashboard_home"),
servers=servers,
app=app
)