202 lines
6.4 KiB
Python
202 lines
6.4 KiB
Python
from app.core.models import App, Port, Server
|
|
from app.core.extensions import db
|
|
from flask import flash
|
|
import re
|
|
|
|
|
|
def validate_app_data(name, server_id, existing_app_id=None):
|
|
"""
|
|
Validate application data
|
|
Returns tuple (valid, error_message)
|
|
"""
|
|
if not name or not server_id:
|
|
return False, "Please fill in all required fields"
|
|
|
|
# Check if app exists with same name on same server
|
|
query = App.query.filter(App.name == name, App.server_id == server_id)
|
|
|
|
# If editing, exclude the current app
|
|
if existing_app_id:
|
|
query = query.filter(App.id != existing_app_id)
|
|
|
|
if query.first():
|
|
return False, f"Application '{name}' already exists on this server"
|
|
|
|
# Find similar app names on this server for suggestion
|
|
similar_apps = App.query.filter(
|
|
App.name.ilike(f"%{name}%"),
|
|
App.server_id == server_id
|
|
).limit(3).all()
|
|
|
|
if similar_apps and (not existing_app_id or not any(app.id == existing_app_id for app in similar_apps)):
|
|
similar_names = ", ".join([f"'{app.name}'" for app in similar_apps])
|
|
return False, f"Similar application names found on this server: {similar_names}"
|
|
|
|
return True, None
|
|
|
|
|
|
def is_port_in_use(port_number, protocol, server_id, exclude_app_id=None):
|
|
"""
|
|
Check if a port+protocol combination is already in use by any application on the server
|
|
"""
|
|
from sqlalchemy import and_
|
|
|
|
# Join App and Port models to find ports used by apps on this server
|
|
query = db.session.query(Port, App).join(App).filter(
|
|
Port.port_number == port_number,
|
|
Port.protocol == protocol,
|
|
App.server_id == server_id
|
|
)
|
|
|
|
# # Exclude the current app if editing
|
|
# if exclude_app_id:
|
|
# query = query.filter(App.id != exclude_app_id)
|
|
|
|
result = query.first()
|
|
|
|
if result:
|
|
return True, result.App.name
|
|
return False, None
|
|
|
|
|
|
def validate_port_data(port_number, protocol, description, server_id=None, app_id=None):
|
|
"""
|
|
Validate port data for an application
|
|
|
|
Args:
|
|
port_number: The port number to validate
|
|
protocol: The protocol (TCP/UDP)
|
|
description: Port description
|
|
server_id: ID of the server
|
|
app_id: ID of the application (for excluding the current app when checking conflicts)
|
|
|
|
Returns:
|
|
Tuple of (valid, clean_port, error_message)
|
|
"""
|
|
# Check if port number is provided
|
|
if not port_number:
|
|
return False, None, "Port number is required"
|
|
|
|
try:
|
|
clean_port = int(port_number)
|
|
except ValueError:
|
|
return False, None, "Port number must be a valid integer"
|
|
|
|
if clean_port < 1 or clean_port > 65535:
|
|
return False, None, "Port number must be between 1 and 65535"
|
|
|
|
# Always check for port conflicts
|
|
in_use, app_name = is_port_in_use(clean_port, protocol, server_id, app_id)
|
|
if in_use:
|
|
return False, clean_port, f"Port {clean_port}/{protocol} is already in use by application '{app_name}'"
|
|
|
|
return True, clean_port, None
|
|
|
|
|
|
def process_app_ports(app_id, port_data, server_id=None):
|
|
"""
|
|
Process port data for an application
|
|
port_data should be a list of tuples (port_number, protocol, description)
|
|
Returns (success, error_message)
|
|
"""
|
|
# Get the app's server_id if not provided
|
|
if not server_id and app_id:
|
|
app = App.query.get(app_id)
|
|
if app:
|
|
server_id = app.server_id
|
|
|
|
# Track the port+protocol combinations we've seen to avoid duplicates
|
|
seen_ports = set()
|
|
|
|
try:
|
|
for port_number, protocol, description in port_data:
|
|
# Skip empty port entries
|
|
if not port_number or not port_number.strip():
|
|
continue
|
|
|
|
port_key = f"{port_number}/{protocol}"
|
|
|
|
# Check for duplicates within this form submission
|
|
if port_key in seen_ports:
|
|
return False, f"Duplicate port: {port_key} is specified multiple times"
|
|
|
|
seen_ports.add(port_key)
|
|
|
|
# Validate the port data
|
|
valid, clean_port, error = validate_port_data(
|
|
port_number, protocol, description, server_id, app_id
|
|
)
|
|
|
|
if not valid:
|
|
return False, error
|
|
|
|
# Check if port already exists for this app
|
|
existing_port = Port.query.filter_by(
|
|
app_id=app_id, port_number=clean_port, protocol=protocol
|
|
).first()
|
|
|
|
if existing_port:
|
|
# Update existing port
|
|
existing_port.description = description
|
|
else:
|
|
# Create new port
|
|
new_port = Port(
|
|
app_id=app_id,
|
|
port_number=clean_port,
|
|
protocol=protocol,
|
|
description=description,
|
|
)
|
|
db.session.add(new_port)
|
|
|
|
return True, None
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return False, str(e)
|
|
|
|
|
|
def save_app(name, server_id, documentation, port_data, app_id=None, url=None):
|
|
"""
|
|
Save or update an application
|
|
Returns (success, app, error_message)
|
|
"""
|
|
try:
|
|
# Validate application data
|
|
valid, error = validate_app_data(name, server_id, app_id)
|
|
if not valid:
|
|
return False, None, error
|
|
|
|
if app_id:
|
|
# Update existing app
|
|
app = App.query.get(app_id)
|
|
if not app:
|
|
return False, None, "Application not found"
|
|
app.name = name
|
|
app.server_id = server_id
|
|
app.documentation = documentation
|
|
app.url = url
|
|
else:
|
|
# Create new app
|
|
app = App(
|
|
name=name,
|
|
server_id=server_id,
|
|
documentation=documentation,
|
|
url=url
|
|
)
|
|
db.session.add(app)
|
|
db.session.flush() # Get the app ID without committing
|
|
|
|
# Remove all existing ports if updating
|
|
if app_id:
|
|
Port.query.filter_by(app_id=app_id).delete()
|
|
|
|
# Process and save ports
|
|
port_success, port_error = process_app_ports(app.id, port_data, server_id)
|
|
if not port_success:
|
|
db.session.rollback()
|
|
return False, None, port_error
|
|
|
|
db.session.commit()
|
|
return True, app, None
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return False, None, str(e)
|