129 lines
4 KiB
Python
129 lines
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"
|
|
|
|
return True, None
|
|
|
|
|
|
def validate_port_data(port_number, protocol, description=None):
|
|
"""
|
|
Validate port data
|
|
Returns tuple (valid, clean_port_number, error_message)
|
|
"""
|
|
# Clean and validate port number
|
|
try:
|
|
port = int(port_number)
|
|
if port < 1 or port > 65535:
|
|
return False, None, "Port must be between 1 and 65535"
|
|
except (ValueError, TypeError):
|
|
return False, None, "Invalid port number"
|
|
|
|
# Validate protocol
|
|
valid_protocols = ["TCP", "UDP", "SCTP", "OTHER"]
|
|
if protocol not in valid_protocols:
|
|
return False, None, f"Protocol must be one of: {', '.join(valid_protocols)}"
|
|
|
|
return True, port, None
|
|
|
|
|
|
def process_app_ports(app_id, port_data):
|
|
"""
|
|
Process port data for an application
|
|
port_data should be a list of tuples (port_number, protocol, description)
|
|
Returns (success, error_message)
|
|
"""
|
|
try:
|
|
for port_number, protocol, description in port_data:
|
|
valid, clean_port, error = validate_port_data(
|
|
port_number, protocol, description
|
|
)
|
|
if not valid:
|
|
continue # Skip invalid ports
|
|
|
|
# Check if port already exists
|
|
existing_port = Port.query.filter_by(
|
|
app_id=app_id, port_number=clean_port
|
|
).first()
|
|
if existing_port:
|
|
# Update existing port
|
|
existing_port.protocol = protocol
|
|
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=None, existing_app_id=None):
|
|
"""
|
|
Save or update an application and its ports
|
|
Returns tuple (success, app_object_or_none, error_message)
|
|
"""
|
|
try:
|
|
# Validate input data
|
|
valid, error = validate_app_data(name, server_id, existing_app_id)
|
|
if not valid:
|
|
return False, None, error
|
|
|
|
# Create or update app
|
|
if existing_app_id:
|
|
# Update existing app
|
|
app = App.query.get(existing_app_id)
|
|
if not app:
|
|
return False, None, f"Application with ID {existing_app_id} not found"
|
|
|
|
app.name = name
|
|
app.server_id = server_id
|
|
app.documentation = documentation
|
|
else:
|
|
# Create new app
|
|
app = App(name=name, server_id=server_id, documentation=documentation)
|
|
db.session.add(app)
|
|
|
|
# Flush to get the app ID if new
|
|
db.session.flush()
|
|
|
|
# Process ports if provided
|
|
if port_data:
|
|
success, error = process_app_ports(app.id, port_data)
|
|
if not success:
|
|
db.session.rollback()
|
|
return False, None, f"Error processing ports: {error}"
|
|
|
|
# Commit all changes
|
|
db.session.commit()
|
|
return True, app, None
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return False, None, str(e)
|