"""WebSocket protocol handlers, sessions, terminal access, and uploads.""" import asyncio import base64 from database_service import database_connection, dict_from_row from datetime import date, datetime, timezone from domain_runtime import domain_manager from domain_service import create_custom_domain_record, create_domain_record, delete_domain_record, get_domain_by_id, get_user_custom_domains, get_user_domains, link_domain_to_server, unlink_domain_from_server, verify_domain_dns import io import json from kubernetes import client from kubernetes.client.rest import ApiException from kubernetes.stream import stream from kubernetes_service import custom_objects_api, k8s_manager, v1 import os import paramiko import pyotp import qrcode import random import re from runtime_config import ( ALLOWED_ORIGINS, CHUNK_UPLOAD_DIR, NODE_IP, PORT, WS_PORT, chunked_uploads, connected_clients, logger, node_sessions, ssh_sessions, ) from server_access import create_server_directory, delete_server_path, download_server_file, exec_server_command, expose_port, get_server_logs, get_server_metrics, read_server_file, read_server_file_base64, reconcile_sftp, remove_port, validate_path_for_server, write_server_file from server_capacity import validate_server_capacity from server_demo import check_demo_eligibility, convert_demo_to_paid from server_lifecycle import delete_server, restart_server, start_server, stop_server from server_provisioning import change_server_version, create_server, extract_server_from_url, extract_token_from_url, generate_sftp_credentials, sync_server_ports_from_k8s, sync_server_status_from_k8s, validate_server_name from server_records import get_all_servers, get_server_access_and_namespace, get_server_by_id, get_user_servers import threading from ticket_service import add_ticket_message, create_ticket, get_all_tickets, get_ticket, get_ticket_stats, get_user_tickets, update_ticket_status import time from typing import Dict import urllib.parse from user_service import add_credits, authenticate_user, create_user, generate_token, generate_two_factor_token, get_full_user_by_id, get_full_user_by_username, get_new_user_credits, get_user_by_id, get_user_by_username, get_user_credits, verify_password, verify_token import uuid import websockets __all__ = ['DateTimeEncoder', 'attach_to_game_console', 'ensure_node_shell_pod', 'execute_minecraft_save_all', 'handle_2fa_disable', 'handle_2fa_enable', 'handle_2fa_login', 'handle_2fa_setup', 'handle_add_credits', 'handle_authenticate', 'handle_change_server_version', 'handle_check_demo_eligibility', 'handle_convert_demo_to_paid', 'handle_copy_path', 'handle_create_backup', 'handle_create_custom_domain', 'handle_create_directory', 'handle_create_domain', 'handle_create_server', 'handle_create_ticket', 'handle_delete_backup', 'handle_delete_custom_domain', 'handle_delete_domain', 'handle_delete_path', 'handle_delete_server', 'handle_download_file', 'handle_exec_command', 'handle_expose_port', 'handle_game_console_command', 'handle_game_console_stream', 'handle_generate_sftp', 'handle_get_backups', 'handle_get_credits', 'handle_get_minecraft_settings', 'handle_get_server_creation_progress', 'handle_get_server_logs', 'handle_get_server_metrics', 'handle_get_servers', 'handle_get_ticket', 'handle_get_ticket_stats', 'handle_get_tickets', 'handle_link_domain', 'handle_list_custom_domains', 'handle_list_domains', 'handle_list_files', 'handle_login', 'handle_move_path', 'handle_node_terminal_connect', 'handle_node_terminal_disconnect', 'handle_node_terminal_input', 'handle_pack_zip', 'handle_read_file', 'handle_register', 'handle_remove_port', 'handle_restart_server', 'handle_restore_backup', 'handle_send_ticket_message', 'handle_start_chunked_upload', 'handle_start_server', 'handle_stop_server', 'handle_sync_server_status', 'handle_unlink_domain', 'handle_unpack_zip', 'handle_update_minecraft_settings', 'handle_update_ticket_status', 'handle_upload_file', 'handle_upload_file_chunk', 'handle_verify_custom_domain', 'handle_vps_terminal_connect', 'handle_vps_terminal_disconnect', 'handle_vps_terminal_input', 'handle_websocket', 'handle_write_file', 'notify_admins', 'notify_ticket_participants', 'parse_ls_output', 'send_message', 'send_verification_email', 'start_websocket_server', 'stream_game_console_output', 'stream_metrics', 'sync_user_domains_with_dns'] async def handle_create_server(client_id, data): """Handle create server request - async version with capacity validation""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] try: # Check if this is a demo server request is_demo = data.get('is_demo', False) # Validate server name first server_name = data.get('name', '').strip() is_valid, error_msg = validate_server_name(server_name) if not is_valid: await send_message(client_id, { 'type': 'error', 'error': error_msg }) return # Add is_demo to config data['is_demo'] = is_demo # First, validate capacity capacity_check = validate_server_capacity(data) if not capacity_check['success']: # Send detailed capacity error await send_message(client_id, { 'type': 'capacity_error', 'error': capacity_check['message'], 'details': capacity_check.get('details', []), 'error_details': capacity_check.get('error_details', []), 'summary': capacity_check.get('summary', {}) }) return # Capacity check passed, proceed with server creation # Create progress callback that sends websocket messages loop = asyncio.get_event_loop() def progress_callback(phase, progress, message): """Thread-safe progress callback that sends websocket messages""" try: # Store progress in client info for reconnection recovery if client_id in connected_clients: connected_clients[client_id]['server_creation_progress'] = { 'phase': phase, 'progress': progress, 'message': message, 'timestamp': time.time() } # Use run_coroutine_threadsafe to send async message from executor thread future = asyncio.run_coroutine_threadsafe( send_message(client_id, { 'type': 'server_creation_progress', 'phase': phase, 'progress': progress, 'message': message }), loop ) # Don't wait for completion to avoid blocking # The future will complete asynchronously except Exception as e: logger.error(f"Error sending progress update: {e}") # Run blocking create_server in executor with progress callback server = await loop.run_in_executor(None, create_server, data, user_id, progress_callback) # Clear server creation progress on completion if client_id in connected_clients: connected_clients[client_id].pop('server_creation_progress', None) # Get user credits user = get_user_by_id(user_id) user_credits = user.get('credits', 0) if user else 0 await send_message(client_id, { 'type': 'server_created', 'server': server, 'credits': user_credits, 'capacity_check': { 'success': True, 'message': capacity_check['message'], 'suitable_nodes': capacity_check.get('suitable_nodes', []), 'summary': capacity_check.get('summary', {}) } }) except Exception as e: logger.error(f"Error creating server: {e}", exc_info=True) # Clear server creation progress on error if client_id in connected_clients: connected_clients[client_id].pop('server_creation_progress', None) # Check if it's a capacity-related error error_msg = str(e) if 'Insufficient capacity' in error_msg or 'No nodes have sufficient capacity' in error_msg: await send_message(client_id, { 'type': 'capacity_error', 'error': error_msg, 'details': [f'Server creation failed: {error_msg}'] }) else: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_check_demo_eligibility(client_id, data): """Handle check demo eligibility request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] try: eligibility = check_demo_eligibility(user_id) await send_message(client_id, { 'type': 'demo_eligibility', 'eligible': eligibility['eligible'], 'has_used_demo': eligibility['has_used_demo'], 'has_purchased_credits': eligibility['has_purchased_credits'], 'reason': eligibility.get('reason') }) except Exception as e: logger.error(f"Error checking demo eligibility: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_convert_demo_to_paid(client_id, data): """Handle convert demo to paid server request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') if not server_id: await send_message(client_id, {'type': 'error', 'error': 'Server ID required'}) return try: result = convert_demo_to_paid(server_id, user_id) # Get updated server and credits server = get_server_by_id(server_id) user = get_user_by_id(user_id) credits = user.get('credits', 0) if user else 0 await send_message(client_id, { 'type': 'demo_converted', 'server': server, 'credits': credits, 'credits_charged': result['credits_charged'], 'new_balance': result['new_balance'] }) except Exception as e: logger.error(f"Error converting demo to paid: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_game_console_command(client_id, data): """Handle game console command""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') command = data.get('command') if not server_id or not command: await send_message(client_id, {'type': 'error', 'error': 'Server ID and command required'}) return try: result = attach_to_game_console(server_id, user_id, command) await send_message(client_id, { 'type': 'game_console_result', 'serverId': server_id, 'command': command, 'result': result }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_game_console_stream(client_id, data): """Handle game console output stream request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') tail_lines = data.get('tailLines', 50) if not server_id: await send_message(client_id, {'type': 'error', 'error': 'Server ID required'}) return try: output = stream_game_console_output(server_id, user_id, tail_lines) await send_message(client_id, { 'type': 'game_console_output', 'serverId': server_id, 'output': output }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_vps_terminal_connect(client_id, data): """Handle VPS SSH terminal connection""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') try: namespace, server = get_server_access_and_namespace(server_id, user_id) if server['type'] != 'vps': await send_message(client_id, {'type': 'error', 'error': 'Not a VPS server'}) return # Find SSH port ssh_port = 22 if server.get('ports'): for p in server['ports']: if p.get('name') == 'ssh' or p.get('internal') == 22: ssh_port = p.get('external', 22) break # Determine credentials # Image mapping for default credentials (fallback) image_map = { 'ubuntu': {'user': 'ubuntu', 'password': 'ubuntu'}, 'debian': {'user': 'debian', 'password': 'debian'}, 'alpine': {'user': 'root', 'password': 'alpine'}, 'centos': {'user': 'centos', 'password': 'centos'} } default_creds = image_map.get(server['image'], image_map['ubuntu']) ssh_user = default_creds['user'] ssh_password = default_creds['password'] # Internal IP for backend to connect (as requested) # We use 192.168.10.2 instead of the node IP for internal bridge access target_ip = "192.168.10.2" await send_message(client_id, { 'type': 'vps_terminal_output', 'serverId': server_id, 'output': f"\r\n\x1b[32m[AnchorHost]\x1b[0m Connecting to {server['ip']} (SSH port {ssh_port})...\r\n" }) # Initialize Paramiko ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: ssh.connect(target_ip, port=ssh_port, username=ssh_user, password=ssh_password, timeout=10) # Open interactive shell channel = ssh.invoke_shell(term='xterm', width=80, height=24) channel.setblocking(0) # Non-blocking for async reading # Store session ssh_sessions[client_id] = { 'client': ssh, 'channel': channel, 'serverId': server_id } await send_message(client_id, { 'type': 'vps_terminal_output', 'serverId': server_id, 'output': "\x1b[32m[AnchorHost]\x1b[0m SSH connection established.\r\n" }) # Start background reading task async def read_ssh_output(): try: while client_id in ssh_sessions and client_id in connected_clients: if channel.recv_ready(): data = channel.recv(4096).decode('utf-8', errors='ignore') if data: await send_message(client_id, { 'type': 'vps_terminal_output', 'serverId': server_id, 'output': data }) await asyncio.sleep(0.05) except Exception as e: logger.error(f"SSH Read Error: {e}") finally: await handle_vps_terminal_disconnect(client_id, {}) asyncio.create_task(read_ssh_output()) except Exception as e: await send_message(client_id, { 'type': 'vps_terminal_output', 'serverId': server_id, 'output': f"\x1b[31m[AnchorHost] Connection failed: {str(e)}\x1b[0m\r\n" }) except Exception as e: logger.error(f"VPS Terminal Error: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_vps_terminal_input(client_id, data): """Handle input from client to SSH channel""" session = ssh_sessions.get(client_id) if not session: return input_data = data.get('input') if input_data: try: session['channel'].send(input_data) except Exception as e: logger.error(f"SSH Send Error: {e}") await handle_vps_terminal_disconnect(client_id, {}) async def handle_vps_terminal_disconnect(client_id, data): """Clean up SSH session""" session = ssh_sessions.pop(client_id, None) if session: try: session['channel'].close() session['client'].close() # Notify frontend if it's still connected if client_id in connected_clients: await send_message(client_id, { 'type': 'vps_terminal_output', 'serverId': session['serverId'], 'output': "\r\n\x1b[31m[AnchorHost] SSH session closed.\x1b[0m\r\n" }) except Exception as e: logger.error(f"SSH Cleanup Error: {e}") async def ensure_node_shell_pod(node_name: str) -> str: """Ensure a privileged shell pod exists on the node in kube-system namespace""" pod_name = f"node-shell-{node_name.lower().replace('.', '-').replace('_', '-')}" namespace = "kube-system" try: pod = v1.read_namespaced_pod(name=pod_name, namespace=namespace) if pod.status.phase == 'Running': return pod_name elif pod.status.phase in ['Failed', 'Succeeded']: v1.delete_namespaced_pod(name=pod_name, namespace=namespace) await asyncio.sleep(1) except Exception: pass # Create the pod pod_manifest = client.V1Pod( metadata=client.V1ObjectMeta( name=pod_name, namespace=namespace, labels={"app": "anchorhost-node-shell", "node": node_name} ), spec=client.V1PodSpec( node_name=node_name, host_pid=True, host_network=True, host_ipc=True, tolerations=[ client.V1Toleration(operator="Exists") ], containers=[ client.V1Container( name="shell", image="alpine:latest", command=["sleep", "86400"], security_context=client.V1SecurityContext(privileged=True), volume_mounts=[ client.V1VolumeMount(name="host-root", mount_path="/host") ] ) ], volumes=[ client.V1Volume( name="host-root", host_path=client.V1HostPathVolumeSource(path="/") ) ], restart_policy="Always" ) ) try: v1.create_namespaced_pod(namespace=namespace, body=pod_manifest) except Exception as e: logger.info(f"Pod creation notice for {pod_name}: {e}") # Wait for pod to be running for _ in range(30): await asyncio.sleep(0.5) try: pod = v1.read_namespaced_pod(name=pod_name, namespace=namespace) if pod.status.phase == 'Running': return pod_name except Exception: pass return pod_name async def handle_node_terminal_connect(client_id, data): """Handle interactive terminal connection to a Kubernetes node""" client_info = connected_clients.get(client_id) if not client_info or not client_info.get('authenticated'): await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] user = get_user_by_id(user_id) if not user or user.get('role') != 'admin': await send_message(client_id, {'type': 'error', 'error': 'Admin access required'}) return node_name = data.get('nodeName') if not node_name: await send_message(client_id, {'type': 'error', 'error': 'Node name required'}) return try: await send_message(client_id, { 'type': 'node_terminal_output', 'nodeName': node_name, 'output': f"\r\n\x1b[32m[AnchorHost]\x1b[0m Initializing node shell on \x1b[36m{node_name}\x1b[0m...\r\n" }) loop = asyncio.get_event_loop() pod_name = await ensure_node_shell_pod(node_name) def create_ws_client(): try: return stream( v1.connect_get_namespaced_pod_exec, pod_name, 'kube-system', container='shell', command=['nsenter', '-t', '1', '-m', '-u', '-i', '-n', '-p', '--', '/bin/sh'], stderr=True, stdin=True, stdout=True, tty=True, _preload_content=False ) except Exception as ns_err: logger.warning(f"nsenter failed ({ns_err}), falling back to chroot: {ns_err}") return stream( v1.connect_get_namespaced_pod_exec, pod_name, 'kube-system', container='shell', command=['chroot', '/host', '/bin/sh'], stderr=True, stdin=True, stdout=True, tty=True, _preload_content=False ) ws_client = await loop.run_in_executor(None, create_ws_client) node_sessions[client_id] = { 'ws_client': ws_client, 'nodeName': node_name } await send_message(client_id, { 'type': 'node_terminal_output', 'nodeName': node_name, 'output': f"\x1b[32m[AnchorHost]\x1b[0m Node terminal connected to \x1b[1m{node_name}\x1b[0m (host root namespace).\r\n\r\n" }) async def read_node_output(): try: while client_id in node_sessions and client_id in connected_clients: session = node_sessions.get(client_id) if not session or not session['ws_client'].is_open(): break def read_from_ws(): if ws_client.is_open(): try: return ws_client.read_stdout(timeout=0.05) except Exception: return None return None output = await loop.run_in_executor(None, read_from_ws) if output: await send_message(client_id, { 'type': 'node_terminal_output', 'nodeName': node_name, 'output': output }) await asyncio.sleep(0.02) except Exception as e: logger.error(f"Node terminal read error: {e}") finally: await handle_node_terminal_disconnect(client_id, {'nodeName': node_name}) asyncio.create_task(read_node_output()) except Exception as e: logger.error(f"Node terminal error: {e}", exc_info=True) await send_message(client_id, { 'type': 'node_terminal_output', 'nodeName': node_name, 'output': f"\x1b[31m[AnchorHost] Connection failed: {str(e)}\x1b[0m\r\n" }) async def handle_node_terminal_input(client_id, data): """Handle input sent to node terminal session""" session = node_sessions.get(client_id) if not session: return input_data = data.get('input') if input_data and session['ws_client'].is_open(): try: loop = asyncio.get_event_loop() await loop.run_in_executor(None, session['ws_client'].write_stdin, input_data) except Exception as e: logger.error(f"Node terminal input write error: {e}") await handle_node_terminal_disconnect(client_id, {}) async def handle_node_terminal_disconnect(client_id, data): """Clean up node terminal session""" session = node_sessions.pop(client_id, None) if session: try: ws_client = session.get('ws_client') if ws_client and ws_client.is_open(): ws_client.close() if client_id in connected_clients: await send_message(client_id, { 'type': 'node_terminal_output', 'nodeName': session.get('nodeName'), 'output': "\r\n\x1b[33m[AnchorHost] Node terminal session closed.\x1b[0m\r\n" }) except Exception as e: logger.error(f"Node terminal cleanup error: {e}") async def handle_websocket(websocket): client_id = str(uuid.uuid4()) # Extract path from websocket object try: path = websocket.request.path if hasattr(websocket, 'request') else getattr(websocket, 'path', '/') except Exception as e: logger.warning(f"Could not extract path from websocket: {e}") path = '/' # Log connection details remote_address = websocket.remote_address if hasattr(websocket, 'remote_address') else 'unknown' safe_path = urllib.parse.urlsplit(path).path logger.info(f"New WebSocket connection: {client_id} from {remote_address}, path: {safe_path}") token = extract_token_from_url(path) server_id = extract_server_from_url(path) user = None if token: user = authenticate_user(token) if user: connected_clients[client_id] = { 'authenticated': True, 'user_id': user['id'], 'username': user['username'], 'role': user.get('role', 'user'), 'websocket': websocket, 'server_id': server_id } await send_message(client_id, { 'type': 'authenticated', 'user': { 'id': user['id'], 'username': user['username'], 'email': user['email'], 'role': user['role'], 'credits': user.get('credits', 0), 'email_verified': user.get('email_verified', False), 'two_factor_enabled': user.get('two_factor_enabled', False) } }) if server_id: # Run sync operations in executor to not block event loop loop = asyncio.get_event_loop() await loop.run_in_executor(None, sync_server_status_from_k8s, server_id, user['id']) await loop.run_in_executor(None, sync_server_ports_from_k8s, server_id, user['id']) asyncio.create_task(stream_metrics(client_id, server_id, user['id'])) else: connected_clients[client_id] = { 'authenticated': False, 'websocket': websocket, 'server_id': server_id } try: async for message in websocket: try: data = json.loads(message) message_type = data.get('type') logger.info(f"📨 Received {message_type} from client {client_id}") # All handlers are already async, just await them if message_type == 'authenticate': await handle_authenticate(client_id, data) elif message_type == 'login': await handle_login(client_id, data) elif message_type == '2fa_login': await handle_2fa_login(client_id, data) elif message_type == '2fa_setup': await handle_2fa_setup(client_id, data) elif message_type == '2fa_enable': await handle_2fa_enable(client_id, data) elif message_type == '2fa_disable': await handle_2fa_disable(client_id, data) elif message_type == 'register': await handle_register(client_id, data) elif message_type == 'get_servers': await handle_get_servers(client_id, data) elif message_type == 'get_server_creation_progress': await handle_get_server_creation_progress(client_id, data) elif message_type == 'create_server': await handle_create_server(client_id, data) elif message_type == 'delete_server': await handle_delete_server(client_id, data) elif message_type == 'start_server': await handle_start_server(client_id, data) elif message_type == 'stop_server': await handle_stop_server(client_id, data) elif message_type == 'restart_server': await handle_restart_server(client_id, data) elif message_type == 'get_server_logs': await handle_get_server_logs(client_id, data) elif message_type == 'get_server_metrics': await handle_get_server_metrics(client_id, data) elif message_type == 'list_files': await handle_list_files(client_id, data) elif message_type == 'read_file': await handle_read_file(client_id, data) elif message_type == 'write_file': await handle_write_file(client_id, data) elif message_type == 'get_credits': await handle_get_credits(client_id, data) elif message_type == 'add_credits': await handle_add_credits(client_id, data) elif message_type == 'expose_port': await handle_expose_port(client_id, data) elif message_type == 'remove_port': await handle_remove_port(client_id, data) elif message_type == 'sync_server_status': await handle_sync_server_status(client_id, data) elif message_type == 'change_server_version': await handle_change_server_version(client_id, data) elif message_type == 'get_minecraft_settings': await handle_get_minecraft_settings(client_id, data) elif message_type == 'update_minecraft_settings': await handle_update_minecraft_settings(client_id, data) elif message_type == 'get_backups': await handle_get_backups(client_id, data) elif message_type == 'create_backup': await handle_create_backup(client_id, data) elif message_type == 'restore_backup': await handle_restore_backup(client_id, data) elif message_type == 'delete_backup': await handle_delete_backup(client_id, data) elif message_type == 'exec_command': await handle_exec_command(client_id, data) elif message_type == 'create_directory': await handle_create_directory(client_id, data) elif message_type == 'delete_path': await handle_delete_path(client_id, data) elif message_type == 'move_path': await handle_move_path(client_id, data) elif message_type == 'copy_path': await handle_copy_path(client_id, data) elif message_type == 'generate_sftp': await handle_generate_sftp(client_id, data) elif message_type == 'get_credits': await handle_pack_zip(client_id, data) elif message_type == 'unpack_zip': await handle_unpack_zip(client_id, data) elif message_type == 'upload_file': await handle_upload_file(client_id, data) elif message_type == 'start_chunked_upload': await handle_start_chunked_upload(client_id, data) elif message_type == 'upload_file_chunk': await handle_upload_file_chunk(client_id, data) elif message_type == 'download_file': await handle_download_file(client_id, data) elif message_type == 'create_ticket': await handle_create_ticket(client_id, data) elif message_type == 'get_tickets': await handle_get_tickets(client_id, data) elif message_type == 'get_ticket': await handle_get_ticket(client_id, data) elif message_type == 'send_ticket_message': await handle_send_ticket_message(client_id, data) elif message_type == 'update_ticket_status': await handle_update_ticket_status(client_id, data) elif message_type == 'get_ticket_stats': await handle_get_ticket_stats(client_id, data) elif message_type == 'game_console_command': await handle_game_console_command(client_id, data) elif message_type == 'game_console_stream': await handle_game_console_stream(client_id, data) elif message_type == 'vps_terminal_connect': await handle_vps_terminal_connect(client_id, data) elif message_type == 'vps_terminal_input': await handle_vps_terminal_input(client_id, data) elif message_type == 'vps_terminal_disconnect': await handle_vps_terminal_disconnect(client_id, data) elif message_type == 'node_terminal_connect': await handle_node_terminal_connect(client_id, data) elif message_type == 'node_terminal_input': await handle_node_terminal_input(client_id, data) elif message_type == 'node_terminal_disconnect': await handle_node_terminal_disconnect(client_id, data) elif message_type == 'create_domain': await handle_create_domain(client_id, data) elif message_type == 'delete_domain': await handle_delete_domain(client_id, data) elif message_type == 'list_domains': await handle_list_domains(client_id, data) elif message_type == 'link_domain': await handle_link_domain(client_id, data) elif message_type == 'unlink_domain': await handle_unlink_domain(client_id, data) elif message_type == 'verify_custom_domain': await handle_verify_custom_domain(client_id, data) elif message_type == 'create_custom_domain': await handle_create_custom_domain(client_id, data) elif message_type == 'list_custom_domains': await handle_list_custom_domains(client_id, data) elif message_type == 'delete_custom_domain': await handle_delete_custom_domain(client_id, data) elif message_type == 'check_demo_eligibility': await handle_check_demo_eligibility(client_id, data) elif message_type == 'convert_demo_to_paid': await handle_convert_demo_to_paid(client_id, data) elif message_type == 'ping': await send_message(client_id, {'type': 'pong'}) else: await send_message(client_id, {'type': 'error', 'error': f'Unknown message type: {message_type}'}) except json.JSONDecodeError: await send_message(client_id, {'type': 'error', 'error': 'Invalid JSON'}) except Exception as e: logger.error(f"Error handling message: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) except websockets.exceptions.ConnectionClosed: logger.info(f"WebSocket connection closed for client {client_id}") finally: # Clean up SSH session if exists await handle_vps_terminal_disconnect(client_id, {}) # Clean up Node terminal session if exists await handle_node_terminal_disconnect(client_id, {}) if client_id in connected_clients: del connected_clients[client_id] # Domain WebSocket handlers # Replace the domain WebSocket handlers in app.py with these fixed versions: # Domain WebSocket handlers """Handle create domain request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] username = client_info['username'] subdomain_name = data.get('subdomain') server_id = data.get('serverId') logger.info(f"DEBUG: Creating domain for user {username}, subdomain: {subdomain_name}") if not subdomain_name: await send_message(client_id, {'type': 'error', 'error': 'Subdomain name required'}) return if not domain_manager: logger.error("Domain manager not initialized!") await send_message(client_id, {'type': 'error', 'error': 'Domain service not available'}) return # Add API token check if not domain_manager.desesec_api_token: logger.warning("Using default API token - this might not work!") try: # Check if subdomain is available availability_result = domain_manager.check_domain_availability(subdomain_name, username) logger.info(f"DEBUG: Availability check: {availability_result}") if not availability_result.get('available', False): await send_message(client_id, {'type': 'error', 'error': availability_result.get('message', 'Domain not available')}) return # Create the subdomain result = domain_manager.create_subdomain(username, subdomain_name) logger.info(f"DEBUG: Create subdomain result: {result}") if result.get('success') and result.get('domain_data'): # Save to database domain_record = { 'user_id': user_id, 'subdomain': result['domain_data']['subdomain'], 'full_domain': result['domain_data']['domain'], 'record_type': result['domain_data']['record_type'], 'ip_address': result['domain_data']['ip_address'], 'ttl': result['domain_data'].get('ttl', 3600), 'server_id': server_id, 'created_at': result['domain_data'].get('created_at', datetime.now().isoformat()) } db_domain = create_domain_record(domain_record) if db_domain: await send_message(client_id, { 'type': 'domain_created', 'domain': db_domain, 'message': result.get('message', 'Domain created successfully') }) else: logger.error("Failed to save domain to database!") await send_message(client_id, { 'type': 'error', 'error': 'Domain created in DNS but failed to save to database' }) else: await send_message(client_id, {'type': 'error', 'error': result.get('message', 'Failed to create domain')}) except Exception as e: logger.error(f"Error creating domain: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_delete_domain(client_id, data): """Handle delete domain request - FIXED VERSION with proper domain type handling""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] username = client_info['username'] domain_id = data.get('domainId') if not domain_id: await send_message(client_id, {'type': 'error', 'error': 'Domain ID required'}) return if not domain_manager: await send_message(client_id, {'type': 'error', 'error': 'Domain service not available'}) return try: # Get domain from database domain = get_domain_by_id(domain_id) if not domain or domain['user_id'] != user_id: await send_message(client_id, {'type': 'error', 'error': 'Domain not found or access denied'}) return # Check if it's a custom domain is_custom = domain.get('is_custom', False) # If domain is linked to a server, unlink it first (this will delete IngressRoute) if domain.get('server_id'): logger.info(f"Domain {domain.get('full_domain') or domain.get('domain')} is linked to server {domain['server_id']}, unlinking first...") # Unlink the domain (this will delete the IngressRoute) if unlink_domain_from_server(domain_id): logger.info(f"✓ Unlinked domain from server and deleted IngressRoute") else: logger.warning(f"⚠ Failed to unlink domain from server, but will continue with deletion") # Delete from DNS ONLY if it's a delegated domain if not is_custom: result = domain_manager.delete_subdomain(username, domain['subdomain']) if not result.get('success'): # Log warning but continue with database deletion logger.warning(f"Failed to delete from DNS: {result.get('message')}") # Delete from database (works for both delegated and custom domains) if delete_domain_record(domain_id): domain_name = domain.get('domain') if is_custom else domain.get('full_domain') await send_message(client_id, { 'type': 'custom_domain_deleted' if is_custom else 'domain_deleted', 'domainId': domain_id, 'message': f'{"Custom domain" if is_custom else "Domain"} {domain_name} deleted successfully' }) else: await send_message(client_id, { 'type': 'error', 'error': 'Failed to delete domain from database' }) except Exception as e: logger.error(f"Error deleting domain: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_list_domains(client_id, data): """Handle list domains request - FIXED to separate custom and delegated domains""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] username = client_info['username'] include_serverless = data.get('includeServerless', True) try: # Get from database - THIS RETURNS ALL DOMAINS (both delegated and custom) all_domains = get_user_domains(user_id, include_serverless) # IMPORTANT: Separate delegated domains from custom domains delegated_domains = [d for d in all_domains if not d.get('is_custom')] # Also sync with DNS to ensure consistency for delegated domains only if domain_manager: # First ensure the user domain exists user_domain = f"u-{username}.{domain_manager.base_domain}" domain_exists = domain_manager._check_domain_exists(user_domain) if domain_exists is None: logger.error(f"Error checking domain {user_domain}") elif not domain_exists: # Create the user domain if it doesn't exist logger.info(f"Creating user domain: {user_domain}") if domain_manager._create_domain(user_domain): logger.info(f"✓ Created user domain: {user_domain}") else: logger.error(f"✗ Failed to create user domain: {user_domain}") # Now list delegated domains from DNS result = domain_manager.list_user_domains(username) if result.get('success'): # Create a set of existing delegated domains from database db_domain_set = {d['full_domain'] for d in delegated_domains} # Add any delegated domains from DNS that aren't in database for dns_domain in result.get('domains', []): if dns_domain['full_domain'] not in db_domain_set: # Add to database domain_record = { 'user_id': user_id, 'subdomain': dns_domain['subdomain'], 'full_domain': dns_domain['full_domain'], 'record_type': dns_domain['record_type'], 'ip_address': dns_domain['ip_address'], 'ttl': dns_domain.get('ttl', 3600), 'created_at': dns_domain.get('created_at', datetime.now().isoformat()) } create_domain_record(domain_record) # Refresh delegated domains list from database all_domains_refreshed = get_user_domains(user_id, include_serverless) delegated_domains = [d for d in all_domains_refreshed if not d.get('is_custom')] # SEND ONLY DELEGATED DOMAINS in the domains_list response # Custom domains should be fetched via list_custom_domains await send_message(client_id, { 'type': 'domains_list', 'domains': delegated_domains, # ONLY delegated domains 'count': len(delegated_domains) }) except Exception as e: logger.error(f"Error listing domains: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_unlink_domain(client_id, data): """Handle unlink domain from server request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] domain_id = data.get('domainId') if not domain_id: await send_message(client_id, {'type': 'error', 'error': 'Domain ID required'}) return try: # Verify domain belongs to user domain = get_domain_by_id(domain_id) if not domain or domain['user_id'] != user_id: await send_message(client_id, {'type': 'error', 'error': 'Domain not found or access denied'}) return # Unlink domain from server if unlink_domain_from_server(domain_id): await send_message(client_id, { 'type': 'domain_unlinked', 'domainId': domain_id, 'message': 'Domain unlinked from server' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to unlink domain'}) except Exception as e: logger.error(f"Error unlinking domain: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_authenticate(client_id, data): token = data.get('token') if not token: await send_message(client_id, {'type': 'auth_error', 'error': 'No token'}) return user = authenticate_user(token) if not user: await send_message(client_id, {'type': 'auth_error', 'error': 'Invalid token'}) return connected_clients[client_id]['authenticated'] = True connected_clients[client_id]['user_id'] = user['id'] connected_clients[client_id]['username'] = user['username'] connected_clients[client_id]['role'] = user.get('role', 'user') await send_message(client_id, { 'type': 'authenticated', 'user': { 'id': user['id'], 'username': user['username'], 'email': user['email'], 'role': user['role'], 'credits': user.get('credits', 0), 'email_verified': user.get('email_verified', False), 'two_factor_enabled': user.get('two_factor_enabled', False) } }) async def handle_login(client_id, data): username = data.get('username') password = data.get('password') if not username or not password: await send_message(client_id, {'type': 'auth_error', 'error': 'Username and password required'}) return user = get_full_user_by_username(username) if not user or not verify_password(password, user['password_hash']): await send_message(client_id, {'type': 'auth_error', 'error': 'Invalid credentials'}) return if user['two_factor_enabled']: # Issue a temporary token and ask for 2FA code temp_token = generate_two_factor_token(user['id']) await send_message(client_id, { 'type': '2fa_required', 'temp_token': temp_token }) return token = generate_token(user['id']) connected_clients[client_id]['authenticated'] = True connected_clients[client_id]['user_id'] = user['id'] connected_clients[client_id]['username'] = user['username'] connected_clients[client_id]['role'] = user.get('role', 'user') await send_message(client_id, { 'type': 'authenticated', 'user': { 'id': user['id'], 'username': user['username'], 'email': user['email'], 'role': user['role'], 'credits': user.get('credits', 0), 'email_verified': user.get('email_verified', False), 'two_factor_enabled': user.get('two_factor_enabled', False) }, 'token': token }) async def handle_2fa_login(client_id, data): temp_token = data.get('temp_token') otp_code = data.get('otp_code') if not temp_token or not otp_code: await send_message(client_id, {'type': 'auth_error', 'error': 'Temporary token and OTP code are required'}) return payload = verify_token(temp_token) if not payload or payload.get('token_type') != 'two_factor': await send_message(client_id, {'type': 'auth_error', 'error': 'Invalid temporary token'}) return user = get_full_user_by_id(payload['userId']) if not user: await send_message(client_id, {'type': 'auth_error', 'error': 'User not found'}) return secret = user.get('two_factor_secret') if not secret: logger.error(f"User {user['id']} missing 2FA secret") await send_message(client_id, {'type': 'auth_error', 'error': '2FA not configured'}) return # Ensure secret is clean secret = str(secret).replace(' ', '').strip() # Ensure OTP code is clean if otp_code: otp_code = str(otp_code).replace(' ', '') try: totp = pyotp.TOTP(secret) current_otp = totp.now() except Exception as e: logger.error(f"Error initializing TOTP for user {user['id']}: {e}") await send_message(client_id, {'type': 'auth_error', 'error': '2FA configuration error'}) return # Using valid_window=4 to allow for +/- 120 seconds clock drift logger.info(f"Verifying OTP for user {user['id']} with server time {datetime.now(timezone.utc)}") logger.info(f"Debug: Server expects OTP: {current_otp} (approx). Provided: {otp_code}") if not totp.verify(otp_code, valid_window=4): logger.warning(f"2FA failed for user {user['id']}. OTP: {otp_code}. Server expected: {current_otp}") await send_message(client_id, {'type': 'auth_error', 'error': 'Invalid OTP code'}) return token = generate_token(user['id']) connected_clients[client_id]['authenticated'] = True connected_clients[client_id]['user_id'] = user['id'] connected_clients[client_id]['username'] = user['username'] connected_clients[client_id]['role'] = user.get('role', 'user') await send_message(client_id, { 'type': 'authenticated', 'user': { 'id': user['id'], 'username': user['username'], 'email': user['email'], 'role': user['role'], 'credits': user.get('credits', 0), 'email_verified': user.get('email_verified', False), 'two_factor_enabled': user.get('two_factor_enabled', False) }, 'token': token }) async def handle_2fa_setup(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] user = get_full_user_by_id(user_id) if not user: await send_message(client_id, {'type': 'error', 'error': 'User not found'}) return # Generate a new secret if one doesn't exist or if re-setup is requested secret = pyotp.random_base32() with database_connection() as conn: cursor = conn.cursor() cursor.execute('UPDATE users SET two_factor_secret = ? WHERE id = ?', (secret, user_id)) conn.commit() provisioning_uri = pyotp.totp.TOTP(secret).provisioning_uri(name=user['email'], issuer_name='AnchorHost') # Generate QR code qr = qrcode.QRCode(version=1, box_size=10, border=5) qr.add_data(provisioning_uri) qr.make(fit=True) img = qr.make_image(fill='black', back_color='white') buffered = io.BytesIO() img.save(buffered) qr_code_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8") await send_message(client_id, { 'type': '2fa_setup_details', 'provisioning_uri': provisioning_uri, 'qr_code': qr_code_base64 }) async def handle_2fa_enable(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] otp_code = data.get('otp_code') if not otp_code: await send_message(client_id, {'type': 'error', 'error': 'OTP code is required'}) return user = get_full_user_by_id(user_id) if not user or not user['two_factor_secret']: await send_message(client_id, {'type': 'error', 'error': '2FA not set up'}) return totp = pyotp.TOTP(user['two_factor_secret']) if not totp.verify(otp_code, valid_window=2): await send_message(client_id, {'type': 'error', 'error': 'Invalid OTP code'}) return with database_connection() as conn: cursor = conn.cursor() cursor.execute('UPDATE users SET two_factor_enabled = True WHERE id = ?', (user_id,)) conn.commit() await send_message(client_id, {'type': '2fa_enabled', 'success': True}) async def handle_2fa_disable(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] otp_code = data.get('otp_code') if not otp_code: await send_message(client_id, {'type': 'error', 'error': 'OTP code is required'}) return user = get_full_user_by_id(user_id) if not user or not user['two_factor_secret']: await send_message(client_id, {'type': 'error', 'error': '2FA not enabled'}) return totp = pyotp.TOTP(user['two_factor_secret']) if not totp.verify(otp_code, valid_window=2): await send_message(client_id, {'type': 'error', 'error': 'Invalid OTP code'}) return with database_connection() as conn: cursor = conn.cursor() cursor.execute('UPDATE users SET two_factor_enabled = False WHERE id = ?', (user_id,)) conn.commit() await send_message(client_id, {'type': '2fa_disabled', 'success': True}) async def handle_register(client_id, data): username = data.get('username') password = data.get('password') email = data.get('email') if not username or not password or not email: await send_message(client_id, {'type': 'auth_error', 'error': 'All fields required'}) return if get_user_by_username(username): await send_message(client_id, {'type': 'auth_error', 'error': 'Username exists'}) return initial_credits = get_new_user_credits() user = create_user(username, email, password, 'user', initial_credits) if not user: await send_message(client_id, {'type': 'auth_error', 'error': 'Failed to create user'}) return if k8s_manager.connected: try: k8s_manager.create_namespace(f"user-{username}".lower()) except Exception: pass token = generate_token(user['id']) connected_clients[client_id]['authenticated'] = True connected_clients[client_id]['user_id'] = user['id'] connected_clients[client_id]['username'] = username await send_message(client_id, { 'type': 'authenticated', 'user': { 'id': user['id'], 'username': username, 'email': email, 'role': 'user', 'credits': 200, 'email_verified': False # ADD THIS LINE - new users are not verified }, 'token': token }) async def handle_get_servers(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] role = client_info.get('role', 'user') # Run blocking DB operation in executor loop = asyncio.get_event_loop() if role == 'admin': servers = await loop.run_in_executor(None, get_all_servers) else: servers = await loop.run_in_executor(None, get_user_servers, user_id) # Proactively reconcile SFTP for all user servers to ensure they are fully working for srv in servers: if srv['type'] != 'vps': threading.Thread(target=reconcile_sftp, args=(srv['id'], user_id), daemon=True).start() # Debug log to verify SFTP fields if servers and len(servers) > 0: first_srv = servers[0] logger.info(f"DEBUG: First server in list: id={first_srv.get('id')}, sftp_user={first_srv.get('sftp_username')}, sftp_port={first_srv.get('sftp_port')}") await send_message(client_id, { 'type': 'servers_list', 'servers': servers, 'count': len(servers) }) async def handle_get_server_creation_progress(client_id, data): """Handle request for current server creation progress""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return # Get stored progress if available progress = client_info.get('server_creation_progress') if progress: # Check if progress is recent (within last 5 minutes) if time.time() - progress.get('timestamp', 0) < 300: await send_message(client_id, { 'type': 'server_creation_progress', 'phase': progress.get('phase', ''), 'progress': progress.get('progress', 0), 'message': progress.get('message', 'Creating server...') }) else: # Progress is stale, clear it client_info.pop('server_creation_progress', None) await send_message(client_id, { 'type': 'server_creation_progress', 'phase': 'Unknown', 'progress': 0, 'message': 'No active server creation found' }) else: # No progress stored await send_message(client_id, { 'type': 'server_creation_progress', 'phase': 'Unknown', 'progress': 0, 'message': 'No active server creation found' }) async def handle_create_domain(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] username = client_info['username'] subdomain_name = data.get('subdomain') server_id = data.get('serverId') logger.info(f"DEBUG: Creating domain for user {username}, subdomain: {subdomain_name}") if not subdomain_name: await send_message(client_id, {'type': 'error', 'error': 'Subdomain name required'}) return if not domain_manager: logger.error("Domain manager not initialized!") await send_message(client_id, {'type': 'error', 'error': 'Domain service not available'}) return # Validate subdomain name if not domain_manager._is_valid_subdomain(subdomain_name): await send_message(client_id, {'type': 'error', 'error': 'Invalid subdomain name. Use only lowercase letters, numbers, and hyphens (1-63 chars)'}) return # Add API token check if not domain_manager.desesec_api_token: logger.warning("Using default API token - this might not work!") try: # Check if subdomain is available (improved check) user_domain = f"u-{username}.{domain_manager.base_domain}" full_subdomain = f"{subdomain_name}.{user_domain}" # Check if domain already exists in database with database_connection() as conn: cursor = conn.cursor() cursor.execute('SELECT id FROM user_domains WHERE full_domain = ? AND user_id = ?', (full_subdomain, user_id)) if cursor.fetchone(): await send_message(client_id, {'type': 'error', 'error': 'Domain already exists in your account'}) return # Check with deSEC API - this will create the user domain if it doesn't exist logger.info(f"Creating domain for user: u-{username}.{domain_manager.base_domain}") # The create_subdomain method in DomainManager already handles creating the user domain # if it doesn't exist, so we just need to call it result = domain_manager.create_subdomain(username, subdomain_name) logger.info(f"DEBUG: Create subdomain result: {result}") if result.get('success') and result.get('domain_data'): # Save to database domain_record = { 'user_id': user_id, 'subdomain': result['domain_data']['subdomain'], 'full_domain': result['domain_data']['domain'], 'record_type': result['domain_data']['record_type'], 'ip_address': result['domain_data']['ip_address'], 'ttl': result['domain_data'].get('ttl', 3600), 'server_id': server_id, 'created_at': result['domain_data'].get('created_at', datetime.now().isoformat()) } db_domain = create_domain_record(domain_record) if db_domain: await send_message(client_id, { 'type': 'domain_created', 'domain': db_domain, 'message': result.get('message', 'Domain created successfully') }) else: logger.error("Failed to save domain to database!") # Try to clean up the DNS record since DB failed domain_manager.delete_subdomain(username, subdomain_name) await send_message(client_id, { 'type': 'error', 'error': 'Failed to save domain to database. Domain creation rolled back.' }) else: await send_message(client_id, {'type': 'error', 'error': result.get('message', 'Failed to create domain')}) except Exception as e: logger.error(f"Error creating domain: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_delete_server(client_id, data): """Handle delete server request - async version""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') if not server_id: await send_message(client_id, {'type': 'error', 'error': 'Server ID required'}) return try: loop = asyncio.get_event_loop() success = await loop.run_in_executor(None, delete_server, server_id, user_id) if success: await send_message(client_id, { 'type': 'server_deleted', 'serverId': server_id, 'credits': get_user_credits(user_id) }) except Exception as e: logger.error(f"Error deleting server: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_start_server(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') try: success = start_server(server_id, user_id) if success: await send_message(client_id, {'type': 'server_updated', 'serverId': server_id, 'status': 'Starting'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_stop_server(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') try: success = stop_server(server_id, user_id) if success: await send_message(client_id, {'type': 'server_updated', 'serverId': server_id, 'status': 'Stopping'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_restart_server(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') if not server_id: await send_message(client_id, {'type': 'error', 'error': 'Server ID required'}) return try: success = restart_server(server_id, user_id) if success: await send_message(client_id, { 'type': 'server_updated', 'serverId': server_id, 'status': 'Restarting' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to restart'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_sync_server_status(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') if not server_id: await send_message(client_id, {'type': 'error', 'error': 'Server ID required'}) return try: sync_server_status_from_k8s(server_id, user_id) sync_server_ports_from_k8s(server_id, user_id) server = get_server_by_id(server_id) await send_message(client_id, { 'type': 'server_updated', 'server': server }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_add_credits(client_id, data): """Handle add credits request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] amount = data.get('amount', 0) if amount <= 0: await send_message(client_id, {'type': 'error', 'error': 'Invalid credit amount'}) return success = add_credits(user_id, amount) if success: await send_message(client_id, { 'type': 'credits_added', 'amount': amount, 'new_balance': get_user_credits(user_id) }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to add credits'}) def sync_user_domains_with_dns(user_id: str) -> bool: """Sync user domains between database and DNS""" try: user = get_user_by_id(user_id) if not user or not domain_manager: return False username = user['username'] # Get domains from DNS dns_result = domain_manager.list_user_domains(username) if not dns_result.get('success'): return False dns_domains = {d['full_domain']: d for d in dns_result.get('domains', [])} # Get domains from database db_domains = get_user_domains(user_id, True) db_domain_set = {d['full_domain'] for d in db_domains} # Add missing DNS domains to database for dns_domain in dns_result.get('domains', []): if dns_domain['full_domain'] not in db_domain_set: domain_record = { 'user_id': user_id, 'subdomain': dns_domain['subdomain'], 'full_domain': dns_domain['full_domain'], 'record_type': dns_domain['record_type'], 'ip_address': dns_domain['ip_address'], 'ttl': dns_domain.get('ttl', 3600), 'created_at': dns_domain.get('created_at', datetime.now().isoformat()) } create_domain_record(domain_record) logger.info(f"Added missing domain to DB: {dns_domain['full_domain']}") return True except Exception as e: logger.error(f"Error syncing domains: {e}") return False async def handle_expose_port(client_id, data): """Handle expose port request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') internal_port = data.get('internalPort') protocol = data.get('protocol', 'TCP') if not server_id or not internal_port: await send_message(client_id, {'type': 'error', 'error': 'Server ID and internal port required'}) return try: success = expose_port(server_id, user_id, internal_port, protocol) if success: await send_message(client_id, { 'type': 'port_exposed', 'serverId': server_id, 'internalPort': internal_port, 'protocol': protocol, 'message': 'Port exposed successfully' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to expose port'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_remove_port(client_id, data): """Handle remove port request - FIXED VERSION""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') internal_port = data.get('internalPort') protocol = data.get('protocol', 'TCP') if not server_id or not internal_port: await send_message(client_id, {'type': 'error', 'error': 'Server ID and internal port required'}) return try: success = remove_port(server_id, user_id, internal_port, protocol) if success: await send_message(client_id, { 'type': 'port_removed', 'serverId': server_id, 'internalPort': internal_port, 'protocol': protocol, 'message': 'Port removed successfully' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to remove port'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_exec_command(client_id, data): """Handle execute command request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') command = data.get('command') if not server_id or not command: await send_message(client_id, {'type': 'error', 'error': 'Server ID and command required'}) return try: result = exec_server_command(server_id, user_id, command) await send_message(client_id, { 'type': 'command_result', 'serverId': server_id, 'command': command, 'result': result }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_create_directory(client_id, data): """Handle create directory request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') dir_path = data.get('dirPath') if not server_id or not dir_path: await send_message(client_id, {'type': 'error', 'error': 'Server ID and directory path required'}) return try: # Validate path for Minecraft servers dir_path = validate_path_for_server(server_id, dir_path) success = create_server_directory(server_id, user_id, dir_path) if success: await send_message(client_id, { 'type': 'directory_created', 'serverId': server_id, 'dirPath': dir_path, 'message': 'Directory created successfully' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to create directory'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_delete_path(client_id, data): """Handle delete path request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') path = data.get('path') if not server_id or not path: await send_message(client_id, {'type': 'error', 'error': 'Server ID and path required'}) return try: # Validate path for Minecraft servers path = validate_path_for_server(server_id, path) success = delete_server_path(server_id, user_id, path) if success: await send_message(client_id, { 'type': 'path_deleted', 'serverId': server_id, 'path': path, 'message': 'Path deleted successfully' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to delete path'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_move_path(client_id, data): """Handle move/rename path request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') src_path = data.get('srcPath') dest_path = data.get('destPath') paths = data.get('paths') # Optional list for batch move if not server_id or (not src_path and not paths) or not dest_path: await send_message(client_id, {'type': 'error', 'error': 'Server ID, source and destination paths required'}) return try: namespace, _ = get_server_access_and_namespace(server_id, user_id) pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: await send_message(client_id, {'type': 'error', 'error': 'Server pod not found'}) return if paths and isinstance(paths, list): # Batch move success_count = 0 errors = [] for p in paths: v_src = validate_path_for_server(server_id, p) v_dest = validate_path_for_server(server_id, f"{dest_path.rstrip('/')}/{p.split('/')[-1]}") if k8s_manager.move_path(namespace, pods[0], v_src, v_dest): success_count += 1 else: errors.append(f"Failed to move {p}") await send_message(client_id, { 'type': 'path_moved', 'serverId': server_id, 'message': f'Moved {success_count} of {len(paths)} items', 'successCount': success_count, 'errors': errors }) else: # Single move src_path = validate_path_for_server(server_id, src_path) dest_path = validate_path_for_server(server_id, dest_path) if k8s_manager.move_path(namespace, pods[0], src_path, dest_path): await send_message(client_id, { 'type': 'path_moved', 'serverId': server_id, 'srcPath': src_path, 'destPath': dest_path, 'message': 'Path moved successfully' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to move path'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_copy_path(client_id, data): """Handle copy path request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') src_path = data.get('srcPath') dest_path = data.get('destPath') paths = data.get('paths') # Optional list for batch copy if not server_id or (not src_path and not paths) or not dest_path: await send_message(client_id, {'type': 'error', 'error': 'Server ID, source and destination paths required'}) return try: namespace, _ = get_server_access_and_namespace(server_id, user_id) pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: await send_message(client_id, {'type': 'error', 'error': 'Server pod not found'}) return if paths and isinstance(paths, list): # Batch copy success_count = 0 errors = [] for p in paths: v_src = validate_path_for_server(server_id, p) v_dest = validate_path_for_server(server_id, f"{dest_path.rstrip('/')}/{p.split('/')[-1]}") if k8s_manager.copy_path(namespace, pods[0], v_src, v_dest): success_count += 1 else: errors.append(f"Failed to copy {p}") await send_message(client_id, { 'type': 'path_copied', 'serverId': server_id, 'message': f'Copied {success_count} of {len(paths)} items', 'successCount': success_count, 'errors': errors }) else: # Single copy src_path = validate_path_for_server(server_id, src_path) dest_path = validate_path_for_server(server_id, dest_path) if k8s_manager.copy_path(namespace, pods[0], src_path, dest_path): await send_message(client_id, { 'type': 'path_copied', 'serverId': server_id, 'srcPath': src_path, 'destPath': dest_path, 'message': 'Path copied successfully' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to copy path'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_pack_zip(client_id, data): """Handle pack directory to zip request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') dir_path = data.get('dirPath') zip_path = data.get('zipPath') if not server_id or not dir_path or not zip_path: await send_message(client_id, {'type': 'zip_error', 'error': 'Server ID, directory path, and zip path required'}) return try: # Validate paths for Minecraft servers dir_path = validate_path_for_server(server_id, dir_path) zip_path = validate_path_for_server(server_id, zip_path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: await send_message(client_id, {'type': 'zip_error', 'error': 'Kubernetes not connected'}) return pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: await send_message(client_id, {'type': 'zip_error', 'error': 'Server pod not found'}) return # Use zip command to pack directory # Escape paths properly for shell import shlex escaped_dir = shlex.quote(dir_path) escaped_zip = shlex.quote(zip_path) dirname = os.path.basename(dir_path.rstrip('/')) escaped_dirname = shlex.quote(dirname) # Create zip file containing directory contents # Try zip, then jar (common in Minecraft), then python3 command = f'cd {escaped_dir} && (zip -r {escaped_zip} . || jar cMf {escaped_zip} . || python3 -m zipfile -c {escaped_zip} .) || (cd {escaped_dir}/.. && (zip -r {escaped_zip} {escaped_dirname} || jar cMf {escaped_zip} {escaped_dirname} || python3 -m zipfile -c {escaped_zip} {escaped_dirname}))' result = k8s_manager.exec_command(namespace, pods[0], ['sh', '-c', command]) # Check if zip file was created check_command = f'test -f {escaped_zip} && echo "exists" || echo "not_found"' check_result = k8s_manager.exec_command(namespace, pods[0], ['sh', '-c', check_command]) if 'exists' in check_result: await send_message(client_id, { 'type': 'zip_packed', 'serverId': server_id, 'dirPath': dir_path, 'zipPath': zip_path, 'message': 'Directory packed successfully' }) else: await send_message(client_id, {'type': 'zip_error', 'error': f'Failed to create zip file. Command output: {result}'}) except Exception as e: logger.error(f"Error packing zip: {e}") await send_message(client_id, {'type': 'zip_error', 'error': str(e)}) async def handle_unpack_zip(client_id, data): """Handle unpack zip file request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') zip_path = data.get('zipPath') extract_path = data.get('extractPath') if not server_id or not zip_path or not extract_path: await send_message(client_id, {'type': 'zip_error', 'error': 'Server ID, zip path, and extract path required'}) return try: # Validate paths for Minecraft servers zip_path = validate_path_for_server(server_id, zip_path) extract_path = validate_path_for_server(server_id, extract_path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: await send_message(client_id, {'type': 'zip_error', 'error': 'Kubernetes not connected'}) return pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: await send_message(client_id, {'type': 'zip_error', 'error': 'Server pod not found'}) return # Use unzip command to extract import shlex escaped_zip = shlex.quote(zip_path) escaped_extract = shlex.quote(extract_path) # Extract zip file to extract_path # Try unzip, then jar, then python3 command = f'unzip -o {escaped_zip} -d {escaped_extract} || (mkdir -p {escaped_extract} && cd {escaped_extract} && jar xf {escaped_zip}) || python3 -m zipfile -e {escaped_zip} {escaped_extract}' result = k8s_manager.exec_command(namespace, pods[0], ['sh', '-c', command]) # Check if extraction was successful (unzip returns 0 on success) # We'll check by verifying files were extracted check_command = f'test -f {escaped_zip} && ls -la {escaped_extract} | head -5' check_result = k8s_manager.exec_command(namespace, pods[0], ['sh', '-c', check_command]) if check_result and ('total' in check_result or len(check_result.strip()) > 0): await send_message(client_id, { 'type': 'zip_unpacked', 'serverId': server_id, 'zipPath': zip_path, 'extractPath': extract_path, 'message': 'Zip file unpacked successfully' }) else: await send_message(client_id, {'type': 'zip_error', 'error': f'Failed to unpack zip file. Command output: {result}'}) except Exception as e: logger.error(f"Error unpacking zip: {e}") await send_message(client_id, {'type': 'zip_error', 'error': str(e)}) async def handle_upload_file(client_id, data): """Handle upload file request (now deprecated - all uploads use chunked upload)""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') file_path = data.get('filePath') file_content = data.get('fileContent') is_base64 = data.get('isBase64', True) if not server_id or not file_path or not file_content: await send_message(client_id, {'type': 'error', 'error': 'Server ID, file path, and content required'}) return try: # Validate path for Minecraft servers file_path = validate_path_for_server(server_id, file_path) # Use binary upload for all files to support all file types user = get_user_by_id(user_id) if not user: await send_message(client_id, {'type': 'error', 'error': 'User not found'}) return if not k8s_manager.connected: await send_message(client_id, { 'type': 'file_uploaded', 'serverId': server_id, 'filePath': file_path, 'message': 'File uploaded successfully' }) return namespace = f"user-{user['username']}".lower() pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: await send_message(client_id, {'type': 'error', 'error': 'Server pod not found'}) return # Use write_file_binary for all files to support binary files success = k8s_manager.write_file_binary(namespace, pods[0], file_path, file_content) if success: await send_message(client_id, { 'type': 'file_uploaded', 'serverId': server_id, 'filePath': file_path, 'message': 'File uploaded successfully' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to upload file'}) except Exception as e: logger.error(f"Error uploading file: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_start_chunked_upload(client_id, data): """Handle start of chunked upload - creates temporary file for disk-based storage""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') file_path = data.get('filePath') total_chunks = data.get('totalChunks') total_size = data.get('totalSize') if not server_id or not file_path or not total_chunks: await send_message(client_id, { 'type': 'upload_error', 'error': 'Server ID, file path, and total chunks required', 'serverId': server_id, 'filePath': file_path }) return try: # Validate path for Minecraft servers file_path = validate_path_for_server(server_id, file_path) # Check access and get namespace namespace, _ = get_server_access_and_namespace(server_id, user_id) # Create unique upload ID # Sanitize file_path to avoid path separator issues in upload_id sanitized_file_path = file_path.replace('/', '_').replace('\\', '_') upload_id = f"{user_id}_{server_id}_{sanitized_file_path}_{int(time.time() * 1000)}" # Create temporary file for storing chunks on disk (avoids memory exhaustion) temp_file_path = os.path.join(CHUNK_UPLOAD_DIR, f"{upload_id}.tmp") # Ensure parent directory exists (should already exist, but be safe) os.makedirs(os.path.dirname(temp_file_path), exist_ok=True) # Initialize upload tracking with disk-based storage chunked_uploads[upload_id] = { 'user_id': user_id, 'server_id': server_id, 'file_path': file_path, 'temp_file': temp_file_path, 'total_chunks': total_chunks, 'total_size': total_size, 'received_chunks': set(), # Track which chunks we've received 'client_id': client_id, 'start_time': time.time() # Track when upload started for cleanup } # Create empty temporary file with open(temp_file_path, 'wb') as f: pass # Create empty file # Send acknowledgment to start sending chunks await send_message(client_id, { 'type': 'upload_started', 'serverId': server_id, 'filePath': file_path, 'uploadId': upload_id }) except Exception as e: logger.error(f"Error starting chunked upload: {e}") await send_message(client_id, { 'type': 'upload_error', 'error': str(e), 'serverId': server_id if 'server_id' in locals() else None, 'filePath': file_path if 'file_path' in locals() else None }) # Clean up temp file if it was created if 'temp_file_path' in locals() and os.path.exists(temp_file_path): try: os.remove(temp_file_path) except: pass async def handle_upload_file_chunk(client_id, data): """Handle a single chunk of a chunked upload - writes chunks to disk to avoid memory exhaustion""" client_info = connected_clients.get(client_id) user_id = client_info['user_id'] if client_info else None server_id = data.get('serverId') file_path = data.get('filePath') if not client_info or not client_info['authenticated']: await send_message(client_id, { 'type': 'upload_error', 'error': 'Not authenticated', 'serverId': server_id, 'filePath': file_path }) return chunk_index = data.get('chunkIndex') chunk_data = data.get('chunkData') total_chunks = data.get('totalChunks') if not server_id or not file_path or chunk_index is None or not chunk_data: await send_message(client_id, { 'type': 'upload_error', 'error': 'Missing required chunk data', 'serverId': server_id, 'filePath': file_path }) return upload_id = None temp_file_path = None try: # Find the upload by matching user_id, server_id, and file_path for uid, upload_info in chunked_uploads.items(): if (upload_info['user_id'] == user_id and upload_info['server_id'] == server_id and upload_info['file_path'] == file_path): upload_id = uid break if not upload_id: await send_message(client_id, { 'type': 'upload_error', 'error': 'Upload session not found', 'serverId': server_id, 'filePath': file_path }) return upload_info = chunked_uploads[upload_id] temp_file_path = upload_info['temp_file'] # Ensure CHUNK_UPLOAD_DIR and parent directory exist (in case they were cleaned up or don't exist) os.makedirs(CHUNK_UPLOAD_DIR, exist_ok=True) parent_dir = os.path.dirname(temp_file_path) if parent_dir != CHUNK_UPLOAD_DIR: # Only create subdirectory if temp_file_path contains subdirectories os.makedirs(parent_dir, exist_ok=True) # Skip if we already received this chunk (idempotent) if chunk_index in upload_info['received_chunks']: logger.warning(f"Chunk {chunk_index} already received for upload {upload_id}, skipping") else: # Decode base64 chunk and write directly to disk file chunk_bytes = base64.b64decode(chunk_data) # Write chunk to temporary file at correct position # We need to write chunks in order, so we'll append and sort later, or use a different approach # For simplicity, we'll store chunks in a sparse file structure # Create file if it doesn't exist before opening with 'r+b' mode if not os.path.exists(temp_file_path): # Create empty file if it doesn't exist try: with open(temp_file_path, 'wb') as f: pass logger.debug(f"Created temp file: {temp_file_path}") except Exception as e: logger.error(f"Failed to create temp file {temp_file_path}: {e}") raise with open(temp_file_path, 'r+b') as f: # Calculate position: assume max chunk size and write at offset # Better approach: use a dictionary file or append with metadata # For now, append chunks with index metadata f.seek(0, 2) # Seek to end # Write chunk index and size before chunk data for reassembly f.write(chunk_index.to_bytes(4, 'big')) f.write(len(chunk_bytes).to_bytes(4, 'big')) f.write(chunk_bytes) upload_info['received_chunks'].add(chunk_index) # Send acknowledgment await send_message(client_id, { 'type': 'upload_chunk_received', 'serverId': server_id, 'filePath': file_path, 'chunkIndex': chunk_index, 'receivedChunks': len(upload_info['received_chunks']), 'totalChunks': total_chunks }) # Check if all chunks received if len(upload_info['received_chunks']) >= upload_info['total_chunks']: # Reassemble file from disk - stream chunks to final file to avoid memory exhaustion logger.info(f"All chunks received for {upload_id}, reassembling file from disk") # Ensure file exists before reading if not os.path.exists(temp_file_path): raise Exception(f"Temporary file not found: {temp_file_path}") # Create final reassembled file path reassembled_file_path = temp_file_path.replace('.tmp', '_reassembled.tmp') try: # Read chunks from disk and write directly to final file in order (streaming approach) # First pass: read all chunk metadata to build index chunk_positions = {} # {chunk_index: (file_position, size)} with open(temp_file_path, 'rb') as f: while True: # Read chunk index index_bytes = f.read(4) if not index_bytes: break chunk_idx = int.from_bytes(index_bytes, 'big') # Read chunk size size_bytes = f.read(4) if not size_bytes: break chunk_size = int.from_bytes(size_bytes, 'big') # Store position where chunk data starts (after index and size bytes) data_start_pos = f.tell() chunk_positions[chunk_idx] = (data_start_pos, chunk_size) # Skip chunk data to next chunk f.seek(chunk_size, 1) # Verify we have all chunks if len(chunk_positions) != upload_info['total_chunks']: raise Exception(f"Missing chunks: expected {upload_info['total_chunks']}, got {len(chunk_positions)}") # Second pass: read chunks in order and write to final file (streaming) # This avoids loading all chunks into memory at once with open(temp_file_path, 'rb') as in_f, open(reassembled_file_path, 'wb') as out_f: for i in range(upload_info['total_chunks']): if i not in chunk_positions: raise Exception(f"Missing chunk {i} in reassembly") file_pos, chunk_size = chunk_positions[i] # Seek to chunk data (skip index and size bytes) in_f.seek(file_pos) # Read and write chunk in smaller pieces to avoid large memory allocation bytes_written = 0 read_chunk_size = 1024 * 1024 # 1MB at a time while bytes_written < chunk_size: read_size = min(read_chunk_size, chunk_size - bytes_written) chunk_data = in_f.read(read_size) if len(chunk_data) != read_size: raise Exception(f"Incomplete chunk {i} read: expected {read_size}, got {len(chunk_data)}") out_f.write(chunk_data) bytes_written += read_size # Clear chunk positions to free memory chunk_positions.clear() # Upload complete file using streaming method to avoid memory issues try: namespace, _ = get_server_access_and_namespace(server_id, user_id) except Exception as access_error: await send_message(client_id, { 'type': 'upload_error', 'error': str(access_error), 'serverId': server_id, 'filePath': file_path }) return pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: await send_message(client_id, { 'type': 'upload_error', 'error': 'Server pod not found', 'serverId': server_id, 'filePath': file_path }) return # Validate path for Minecraft servers before uploading file_path = validate_path_for_server(server_id, file_path) # Upload the file using streaming method (reads from disk, doesn't load into memory) success = k8s_manager.write_file_binary_from_disk(namespace, pods[0], file_path, reassembled_file_path) if success: await send_message(client_id, { 'type': 'file_uploaded', 'serverId': server_id, 'filePath': file_path, 'message': 'File uploaded successfully' }) logger.info(f"Successfully uploaded file {file_path} for upload {upload_id}") else: await send_message(client_id, { 'type': 'upload_error', 'error': 'Failed to upload file', 'serverId': server_id, 'filePath': file_path }) finally: # Clean up temporary files and upload tracking try: if temp_file_path and os.path.exists(temp_file_path): os.remove(temp_file_path) except Exception as cleanup_error: logger.warning(f"Error cleaning up temp file {temp_file_path}: {cleanup_error}") try: if reassembled_file_path and os.path.exists(reassembled_file_path): os.remove(reassembled_file_path) except Exception as cleanup_error: logger.warning(f"Error cleaning up reassembled file {reassembled_file_path}: {cleanup_error}") del chunked_uploads[upload_id] except Exception as e: logger.error(f"Error handling upload chunk: {e}", exc_info=True) await send_message(client_id, { 'type': 'upload_error', 'error': str(e), 'serverId': server_id if 'server_id' in locals() else None, 'filePath': file_path if 'file_path' in locals() else None }) # Clean up on error if upload_id and upload_id in chunked_uploads: temp_file = chunked_uploads[upload_id].get('temp_file') try: if temp_file and os.path.exists(temp_file): os.remove(temp_file) except: pass del chunked_uploads[upload_id] async def handle_download_file(client_id, data): """Handle download file request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') file_path = data.get('filePath') if not server_id or not file_path: await send_message(client_id, {'type': 'error', 'error': 'Server ID and file path required'}) return try: content = download_server_file(server_id, user_id, file_path) await send_message(client_id, { 'type': 'file_downloaded', 'serverId': server_id, 'filePath': file_path, 'fileContent': content, 'isBase64': True }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_list_files(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') path = data.get('path', '/') try: # Validate path for Minecraft servers path = validate_path_for_server(server_id, path) # Use exec_command instead of direct file listing for better compatibility result = exec_server_command(server_id, user_id, f'ls -la "{path}" 2>/dev/null || echo "no_access"') # Parse the result and convert to files list files = parse_ls_output(result, path) await send_message(client_id, { 'type': 'files_list', 'serverId': server_id, 'path': path, 'files': files }) except Exception as e: logger.error(f"Error listing files: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) def parse_ls_output(output, path): """Parse ls -la output and convert to file list format""" files = [] lines = output.strip().split('\n') for line in lines: # Skip the "total X" line and empty lines if line.startswith('total ') or not line.strip(): continue try: parts = line.split() if len(parts) < 9: continue permissions = parts[0] name = ' '.join(parts[8:]) if name in ['.', '..']: continue file_type = "directory" if permissions.startswith('d') else "file" # Get file size size = 0 try: size = int(parts[4]) except (ValueError, IndexError): pass # Parse modification time try: # Format: "Jan 01 12:00" or "Jan 01 2023" date_parts = parts[5:8] if len(date_parts) == 3: month, day, time_or_year = date_parts if ':' in time_or_year: # Current year, use current year year = datetime.now().year modified_str = f"{month} {day} {year} {time_or_year}" modified = datetime.strptime(modified_str, "%b %d %Y %H:%M") else: # Different year year = time_or_year modified_str = f"{month} {day} {year}" modified = datetime.strptime(modified_str, "%b %d %Y") else: modified = datetime.now() except Exception as e: logger.warning(f"Error parsing date: {e}") modified = datetime.now() files.append({ "name": name, "type": file_type, "size": size, "modified": modified.isoformat() }) except Exception as e: logger.warning(f"Error parsing file entry: {line}, error: {e}") continue return files async def handle_read_file(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') file_path = data.get('filePath') try: # Check if file is an image by extension image_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp', '.svg', '.ico'] is_image = any(file_path.lower().endswith(ext) for ext in image_extensions) if is_image: content = read_server_file_base64(server_id, user_id, file_path) await send_message(client_id, { 'type': 'file_content', 'serverId': server_id, 'filePath': file_path, 'content': content, 'isBase64': True, 'isImage': True }) else: content = read_server_file(server_id, user_id, file_path) await send_message(client_id, { 'type': 'file_content', 'serverId': server_id, 'filePath': file_path, 'content': content }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_get_minecraft_settings(client_id, data): """Get Minecraft server.properties settings""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') try: content = read_server_file(server_id, user_id, 'server.properties') if not content: # Try /data/server.properties if default fails content = read_server_file(server_id, user_id, '/data/server.properties') settings = {} if content: for line in content.splitlines(): line = line.strip() if line and not line.startswith('#') and '=' in line: key, value = line.split('=', 1) settings[key.strip()] = value.strip() await send_message(client_id, { 'type': 'minecraft_settings', 'serverId': server_id, 'settings': settings }) except Exception as e: logger.error(f"Error getting Minecraft settings: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_update_minecraft_settings(client_id, data): """Update Minecraft server.properties settings""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') updates = data.get('settings', {}) try: # 1. Read existing content file_path = '/data/server.properties' content = read_server_file(server_id, user_id, file_path) if not content: # Try server.properties if /data/ fails (though validate_path should handle it) file_path = 'server.properties' content = read_server_file(server_id, user_id, file_path) if not content: # If still no content, create a new one or fail content = "# Minecraft server properties\n" file_path = '/data/server.properties' # 2. Parse and update lines = content.splitlines() new_lines = [] updated_keys = set() for line in lines: trimmed = line.strip() if trimmed and not trimmed.startswith('#') and '=' in trimmed: key = trimmed.split('=', 1)[0].strip() if key in updates: new_lines.append(f"{key}={updates[key]}") updated_keys.add(key) else: new_lines.append(line) else: new_lines.append(line) # 3. Add any keys that weren't in the original file for key, value in updates.items(): if key not in updated_keys: new_lines.append(f"{key}={value}") # 4. Write back new_content = "\n".join(new_lines) success = write_server_file(server_id, user_id, file_path, new_content) if success: await send_message(client_id, { 'type': 'minecraft_settings_updated', 'serverId': server_id }) # Optionally restart server to apply changes? # User might want to change multiple things before restarting. else: await send_message(client_id, {'type': 'error', 'error': 'Failed to save settings'}) except Exception as e: logger.error(f"Error updating Minecraft settings: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_get_backups(client_id, data): """List backups for a server""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return server_id = data.get('serverId') if not server_id: await send_message(client_id, {'type': 'error', 'error': 'Server ID required'}) return try: with database_connection() as conn: cursor = conn.cursor() cursor.execute('SELECT * FROM backups WHERE server_id = ? ORDER BY created_at DESC', (server_id,)) rows = cursor.fetchall() backups = [dict_from_row(row) for row in rows] await send_message(client_id, { 'type': 'backups_list', 'serverId': server_id, 'backups': backups }) except Exception as e: logger.error(f"Error getting backups: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_create_backup(client_id, data): """Trigger a new backup for a server""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') backup_name = data.get('name', f"Manual Backup {datetime.now().strftime('%Y-%m-%d %H:%M')}") if not server_id: await send_message(client_id, {'type': 'error', 'error': 'Server ID required'}) return try: # Get server info server = get_server_by_id(server_id) if not server: raise Exception("Server not found") backup_id = str(uuid.uuid4()) # Insert into DB with database_connection() as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO backups (id, server_id, name, type, status, created_at) VALUES (?, ?, ?, ?, ?, ?) ''', (backup_id, server_id, backup_name, 'Manual', 'In Progress', datetime.now())) conn.commit() # Notify client await send_message(client_id, { 'type': 'backup_status_updated', 'serverId': server_id, 'backup': { 'id': backup_id, 'status': 'In Progress', 'name': backup_name } }) # Start backup in background thread/task async def run_backup(): try: server_type = server.get('type') image = server.get('image') k8s_name = server.get('k8s_name') namespace = server.get('namespace') # 1. Minecraft Consistency (RCON) if image == 'minecraft': await execute_minecraft_save_all(server_id) # 2. Identify the PVC and Volume # PVC is {server_id}-data for app/game servers, {server_id}-rootdisk for VPS pvc_name = f"{server_id}-data" if server_type == 'vps': pvc_name = f"{server_id}-rootdisk" # Get PVC to find volumeName pvc = v1.read_namespaced_persistent_volume_claim(pvc_name, namespace) volume_name = pvc.spec.volume_name if not volume_name: raise Exception(f"Could not find volume for PVC {pvc_name}") # 3. Create Longhorn Snapshot snapshot_obj = { "apiVersion": "longhorn.io/v1beta2", "kind": "Snapshot", "metadata": { "name": backup_id, "namespace": "longhorn-system", "labels": { "app": server_id, "backup-id": backup_id } }, "spec": { "volume": volume_name, "createSnapshot": True } } custom_objects_api.create_namespaced_custom_object( group="longhorn.io", version="v1beta2", namespace="longhorn-system", plural="snapshots", body=snapshot_obj ) # Wait for snapshot to be ready logger.info(f"Waiting for Longhorn snapshot {backup_id} to be ready...") snapshot_ready = False size = "Unknown" for _ in range(30): # 60 seconds max await asyncio.sleep(2) snap = custom_objects_api.get_namespaced_custom_object( group="longhorn.io", version="v1beta2", namespace="longhorn-system", plural="snapshots", name=backup_id ) status = snap.get('status', {}) if status and status.get('readyToUse'): snapshot_ready = True # Use restoreSize (full volume size) instead of incremental size size = status.get('restoreSize', 'Unknown') logger.info(f"Longhorn snapshot {backup_id} is ready. Restore Size: {size}") break if not snapshot_ready: raise Exception(f"Longhorn snapshot {backup_id} did not become ready in time") # 4. Update DB with database_connection() as conn: cursor = conn.cursor() cursor.execute(''' UPDATE backups SET status = ?, completed_at = ?, size = ? WHERE id = ? ''', ('Completed', datetime.now(), size, backup_id)) conn.commit() await send_message(client_id, { 'type': 'backup_status_updated', 'serverId': server_id, 'backup': { 'id': backup_id, 'status': 'Completed', 'size': size } }) except Exception as e: logger.error(f"Backup failed for {server_id}: {e}") with database_connection() as conn: cursor = conn.cursor() cursor.execute('UPDATE backups SET status = ? WHERE id = ?', ('Failed', backup_id)) conn.commit() await send_message(client_id, { 'type': 'error', 'error': f"Backup failed: {str(e)}" }) asyncio.create_task(run_backup()) except Exception as e: logger.error(f"Error triggering backup: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def execute_minecraft_save_all(server_id): """Send save-all command via RCON if server is running""" try: # Get server port from ports JSON server = get_server_by_id(server_id) ports = json.loads(server['ports']) rcon_port = next((p['external'] for p in ports if p['name'] == 'rcon'), None) if rcon_port: from mcrcon import MCRcon # Get RCON password from environment or common secret rcon_pass = os.environ.get('MINECRAFT_RCON_PASSWORD', 'change-me-in-production') with MCRcon(NODE_IP, rcon_pass, port=rcon_port) as mcr: mcr.command("/save-off") mcr.command("/save-all") # Small delay to ensure save is complete await asyncio.sleep(2) mcr.command("/save-on") except Exception as e: logger.warning(f"Failed to coordinate Minecraft save for {server_id}: {e}") async def handle_delete_backup(client_id, data): """Delete a backup""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return backup_id = data.get('backupId') server_id = data.get('serverId') user_id = client_info['user_id'] if not backup_id or not server_id: await send_message(client_id, {'type': 'error', 'error': 'Backup ID and Server ID required'}) return try: # Get namespace namespace, _ = get_server_access_and_namespace(server_id, user_id) # Delete from Kubernetes (Longhorn Snapshot) try: custom_objects_api.delete_namespaced_custom_object( group="longhorn.io", version="v1beta2", namespace="longhorn-system", plural="snapshots", name=backup_id ) logger.info(f"✓ Deleted Longhorn Snapshot {backup_id}") except Exception as e: logger.warning(f"Could not delete Longhorn Snapshot {backup_id}: {e}") with database_connection() as conn: cursor = conn.cursor() cursor.execute('DELETE FROM backups WHERE id = ?', (backup_id,)) conn.commit() await send_message(client_id, { 'type': 'backup_deleted', 'serverId': server_id, 'backupId': backup_id }) except Exception as e: logger.error(f"Error deleting backup: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_restore_backup(client_id, data): """Restore a backup to a server""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return backup_id = data.get('backupId') server_id = data.get('serverId') user_id = client_info['user_id'] if not backup_id or not server_id: await send_message(client_id, {'type': 'error', 'error': 'Backup ID and Server ID required'}) return try: namespace, server = get_server_access_and_namespace(server_id, user_id) await send_message(client_id, { 'type': 'info', 'message': 'Restoration process started. Stopping server...' }) # 1. Stop server stop_server(server_id, user_id) if k8s_manager.connected: # Wait for pods to be gone (Wait up to 60s) logger.info(f"Waiting for pods of {server_id} to stop...") for i in range(12): pods = v1.list_namespaced_pod(namespace, label_selector=f"app={server_id}") if not pods.items: break logger.info(f" ... {len(pods.items)} pods still running") await asyncio.sleep(5) # 2. Handle PVC restoration pvc_name = f"{server_id}-data" if server['type'] == 'vps': pvc_name = f"{server_id}-rootdisk" try: # Get old PVC spec to preserve storage size and storage class old_pvc = v1.read_namespaced_persistent_volume_claim(pvc_name, namespace) storage_size = old_pvc.spec.resources.requests['storage'] storage_class = old_pvc.spec.storage_class_name await send_message(client_id, { 'type': 'info', 'message': 'Recreating volume from backup...' }) # Delete old PVC v1.delete_namespaced_persistent_volume_claim(pvc_name, namespace) # Wait for PVC to be gone for _ in range(12): try: v1.read_namespaced_persistent_volume_claim(pvc_name, namespace) await asyncio.sleep(5) except ApiException as e: if e.status == 404: break # 3. Create new PVC from backup pvc_body = client.V1PersistentVolumeClaim( metadata=client.V1ObjectMeta(name=pvc_name), spec=client.V1PersistentVolumeClaimSpec( access_modes=["ReadWriteOnce"], resources=client.V1ResourceRequirements( requests={"storage": storage_size} ), storage_class_name=storage_class, data_source=client.V1TypedLocalObjectReference( api_group="longhorn.io", kind="Snapshot", name=backup_id ) ) ) v1.create_namespaced_persistent_volume_claim(namespace=namespace, body=pvc_body) logger.info(f"✓ Created new PVC {pvc_name} from backup {backup_id}") except ApiException as e: if e.status == 404: raise Exception(f"PVC {pvc_name} not found") raise e else: await asyncio.sleep(2) # Mock delay # 4. Start server start_server(server_id, user_id) await send_message(client_id, { 'type': 'info', 'message': 'Restoration completed successfully. Server is starting.' }) logger.info(f"✓ Successfully restored backup {backup_id} for server {server_id}") except Exception as e: logger.error(f"Error restoring backup: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_change_server_version(client_id, data): """Handle change server version request - free of charge, billing date unchanged""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') new_version = data.get('version') java_version = data.get('java_version') loader = data.get('loader') if not server_id or not new_version: await send_message(client_id, {'type': 'error', 'error': 'Server ID and version required'}) return try: # Run blocking change_server_version in executor loop = asyncio.get_event_loop() # Use lambda or functools.partial to pass keyword arguments if run_in_executor doesn't support them directly in this context # But run_in_executor supports *args, not **kwargs directly for the function. # We need to wrap it. import functools func = functools.partial(change_server_version, server_id, user_id, new_version, java_version=java_version, loader=loader) success = await loop.run_in_executor(None, func) if success: # Get updated server info server = get_server_by_id(server_id) await send_message(client_id, { 'type': 'server_version_changed', 'serverId': server_id, 'version': new_version, 'server': server }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to change server version'}) except Exception as e: logger.error(f"Error changing server version: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_write_file(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') file_path = data.get('filePath') content = data.get('content') try: success = write_server_file(server_id, user_id, file_path, content) if success: await send_message(client_id, {'type': 'file_written', 'serverId': server_id, 'filePath': file_path}) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to write file'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_generate_sftp(client_id, data): """Generate SFTP credentials for a server""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: return user_id = client_info['user_id'] server_id = data.get('serverId') try: namespace, server = get_server_access_and_namespace(server_id, user_id) # Only for non-VPS if server['type'] == 'vps': await send_message(client_id, {'type': 'error', 'error': 'SFTP sidecar not supported for VPS'}) return u, p, port = generate_sftp_credentials() with database_connection() as conn: cursor = conn.cursor() cursor.execute( 'UPDATE servers SET sftp_username = ?, sftp_password = ?, sftp_port = ? WHERE id = ?', (u, p, port, server_id) ) conn.commit() # Reconcile K8s resources reconcile_sftp(server_id, user_id) await send_message(client_id, { 'type': 'sftp_generated', 'serverId': server_id, 'sftp_username': u, 'sftp_password': p, 'sftp_port': port }) logger.info(f"✓ Generated SFTP credentials for server {server_id}") except Exception as e: logger.error(f"Error generating SFTP: {e}") await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_get_credits(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] try: credits = get_user_credits(user_id) await send_message(client_id, {'type': 'credits_balance', 'credits': credits}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_get_server_logs(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') try: logs = get_server_logs(server_id, user_id) await send_message(client_id, {'type': 'server_logs', 'serverId': server_id, 'logs': logs}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_get_server_metrics(client_id, data): client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] server_id = data.get('serverId') try: metrics = get_server_metrics(server_id, user_id) # Proactively reconcile SFTP if needed # This ensures that even old servers get SFTP sidecar eventually # We only do this occasionally or if metrics are successful if metrics and random.random() < 0.1: # 10% chance to reconcile on metrics call to avoid overload reconcile_sftp(server_id, user_id) await send_message(client_id, {'type': 'server_metrics', 'serverId': server_id, 'metrics': metrics}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_create_ticket(client_id, data): """Handle create ticket request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] subject = data.get('subject') description = data.get('description') priority = data.get('priority', 'medium') category = data.get('category', 'general') if not subject or not description: await send_message(client_id, {'type': 'error', 'error': 'Subject and description required'}) return try: ticket = create_ticket(user_id, subject, description, priority, category) if ticket: await send_message(client_id, { 'type': 'ticket_created', 'ticket': ticket }) # Notify admins about new ticket await notify_admins('new_ticket', ticket) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to create ticket'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_get_tickets(client_id, data): """Handle get tickets request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] role = client_info.get('role', 'user') status_filter = data.get('status_filter') priority_filter = data.get('priority_filter') try: if role == 'admin': tickets = get_all_tickets(status_filter, priority_filter) else: include_closed = data.get('include_closed', False) tickets = get_user_tickets(user_id, include_closed) await send_message(client_id, { 'type': 'tickets_list', 'tickets': tickets }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_get_ticket(client_id, data): """Handle get ticket request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return ticket_id = data.get('ticket_id') if not ticket_id: await send_message(client_id, {'type': 'error', 'error': 'Ticket ID required'}) return try: ticket = get_ticket(ticket_id) if ticket: # Check if user has access to this ticket user_id = client_info['user_id'] role = client_info.get('role', 'user') if role != 'admin' and str(ticket.get('user_id')) != str(user_id): await send_message(client_id, {'type': 'error', 'error': 'Access denied'}) return await send_message(client_id, { 'type': 'ticket_details', 'ticket': ticket }) else: await send_message(client_id, {'type': 'error', 'error': 'Ticket not found'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_send_ticket_message(client_id, data): """Handle send ticket message request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] role = client_info.get('role', 'user') ticket_id = data.get('ticket_id') message = data.get('message') if not ticket_id or not message: await send_message(client_id, {'type': 'error', 'error': 'Ticket ID and message required'}) return try: # Verify ticket access ticket = get_ticket(ticket_id) if not ticket: await send_message(client_id, {'type': 'error', 'error': 'Ticket not found'}) return if role != 'admin' and str(ticket.get('user_id')) != str(user_id): await send_message(client_id, {'type': 'error', 'error': 'Access denied'}) return # Add message is_admin = (role == 'admin') message_data = add_ticket_message(ticket_id, user_id, message, is_admin) if message_data: await send_message(client_id, { 'type': 'ticket_message_sent', 'message': message_data }) # Notify other participants await notify_ticket_participants(ticket_id, 'new_message', { 'ticket_id': ticket_id, 'message': message_data }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to send message'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_update_ticket_status(client_id, data): """Handle update ticket status request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return role = client_info.get('role', 'user') if role != 'admin': await send_message(client_id, {'type': 'error', 'error': 'Admin access required'}) return ticket_id = data.get('ticket_id') status = data.get('status') if not ticket_id or not status: await send_message(client_id, {'type': 'error', 'error': 'Ticket ID and status required'}) return try: success = update_ticket_status(ticket_id, status, client_info['user_id']) if success: await send_message(client_id, { 'type': 'ticket_status_updated', 'ticket_id': ticket_id, 'status': status }) # Notify ticket owner about status change await notify_ticket_participants(ticket_id, 'status_updated', { 'ticket_id': ticket_id, 'status': status, 'updated_by': client_info['username'] }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to update ticket status'}) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_get_ticket_stats(client_id, data): """Handle get ticket statistics request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return try: stats = get_ticket_stats() await send_message(client_id, { 'type': 'ticket_stats', 'stats': stats }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def notify_admins(event_type: str, data: Dict): """Notify all connected admins about an event""" for client_id, client_info in connected_clients.items(): if client_info.get('authenticated') and client_info.get('role') == 'admin': try: await send_message(client_id, { 'type': f'admin_notification_{event_type}', 'data': data }) except Exception as e: logger.error(f"Error notifying admin {client_id}: {e}") async def notify_ticket_participants(ticket_id: str, event_type: str, data: Dict): """Notify all participants of a ticket about an event""" ticket = get_ticket(ticket_id) if not ticket: return participants = set() participants.add(ticket['user_id']) # Ticket owner # Add all users who sent messages in the ticket for message in ticket.get('messages', []): participants.add(message['user_id']) for client_id, client_info in connected_clients.items(): if client_info.get('authenticated') and client_info.get('user_id') in participants: try: await send_message(client_id, { 'type': f'ticket_{event_type}', 'data': data }) except Exception as e: logger.error(f"Error notifying ticket participant {client_id}: {e}") async def stream_metrics(client_id: str, server_id: str, user_id: str): while client_id in connected_clients and connected_clients[client_id].get('authenticated'): try: metrics = get_server_metrics(server_id, user_id) await send_message(client_id, { 'type': 'server_metrics', 'serverId': server_id, 'metrics': metrics }) await asyncio.sleep(1) except Exception: await asyncio.sleep(5) async def send_message(client_id, message): """Send message to WebSocket client with proper error handling""" if client_id not in connected_clients: logger.warning(f"Cannot send message to {client_id}: client not connected") return False try: websocket = connected_clients[client_id]['websocket'] # Use custom encoder to handle datetime objects message_json = json.dumps(message, cls=DateTimeEncoder) # Log outgoing messages for debugging message_type = message.get('type', 'unknown') logger.debug(f"📤 Sending {message_type} to {client_id}") await websocket.send(message_json) logger.debug(f"✓ Successfully sent {message_type} to {client_id}") return True except websockets.exceptions.ConnectionClosed as e: logger.warning(f"Connection closed while sending to {client_id}: {e}") if client_id in connected_clients: del connected_clients[client_id] return False except TypeError as e: logger.error(f"JSON serialization error for {client_id}: {e}") logger.error(f"Problematic message: {message}") if client_id in connected_clients: del connected_clients[client_id] return False except Exception as e: logger.error(f"Error sending message to {client_id}: {e}", exc_info=True) if client_id in connected_clients: del connected_clients[client_id] return False def send_verification_email(email: str, token: str): """Send verification email (placeholder implementation)""" try: verification_url = f"http://localhost:{PORT}/verify-email?token={token}" logger.info(f"Verification email would be sent to {email} with token {token}") logger.info(f"Verification URL: {verification_url}") # TODO: Implement actual email sending service except Exception as e: logger.error(f"Error sending verification email: {e}") def start_websocket_server(): """Start WebSocket server in a separate thread with proper event loop""" def run_server(): logger.info(f"Starting WebSocket server on port {WS_PORT}...") # Create a new event loop for this thread loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) async def start_server(): """Async function to start the WebSocket server""" try: server = await websockets.serve( handle_websocket, "0.0.0.0", WS_PORT, origins=[None, *ALLOWED_ORIGINS], ping_interval=20, # Send ping every 20 seconds ping_timeout=60, # Wait up to 60 seconds for pong max_size=10 * 1024 * 1024, # 10MB max message size for file uploads compression=None # Disable compression for better performance ) logger.info(f"✓ WebSocket server started successfully on 0.0.0.0:{WS_PORT}") logger.info(f" Listening for connections...") # Keep the server running await asyncio.Future() # run forever except OSError as e: if e.errno == 98 or "Address already in use" in str(e): logger.error(f"❌ Port {WS_PORT} is already in use!") logger.error(f" Kill the process using: lsof -ti:{WS_PORT} | xargs kill -9") else: logger.error(f"❌ WebSocket server OS error: {e}") except Exception as e: logger.error(f"❌ WebSocket server failed to start: {e}") logger.error(f" Error type: {type(e).__name__}") import traceback logger.error(traceback.format_exc()) try: # Run the server loop.run_until_complete(start_server()) except KeyboardInterrupt: logger.info("WebSocket server shutting down...") except Exception as e: logger.error(f"WebSocket server error: {e}") finally: # Clean up pending = asyncio.all_tasks(loop) for task in pending: task.cancel() loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) loop.close() logger.info("WebSocket server stopped") thread = threading.Thread(target=run_server, daemon=True, name="WebSocketServer") thread.start() return thread async def handle_link_domain(client_id, data): """Handle link domain to server request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] username = client_info['username'] domain_id = data.get('domainId') server_id = data.get('serverId') if not domain_id: await send_message(client_id, {'type': 'error', 'error': 'Domain ID required'}) return logger.info(f"DEBUG: Linking domain {domain_id} to server {server_id} for user {user_id}") try: # Verify domain belongs to user domain = get_domain_by_id(domain_id) if not domain or domain['user_id'] != user_id: logger.error(f"Domain not found or access denied. Domain: {domain}, user_id: {user_id}") await send_message(client_id, {'type': 'error', 'error': 'Domain not found or access denied'}) return # If server_id is None or empty string, unlink if not server_id or server_id == '': if unlink_domain_from_server(domain_id): await send_message(client_id, { 'type': 'domain_unlinked', 'domainId': domain_id, 'message': 'Domain unlinked from server and IngressRoute removed' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to unlink domain'}) return # Verify server belongs to user or admin server = get_server_by_id(server_id) if not server: logger.error(f"Server not found: {server_id}") await send_message(client_id, {'type': 'error', 'error': 'Server not found'}) return requester = get_user_by_id(user_id) server_owner_id = str(server.get('owner_id') or server.get('ownerId', '')) req_id = str(user_id) if server_owner_id != req_id and (not requester or requester.get('role') != 'admin'): logger.error(f"Server access denied. Server owner: {server_owner_id}, User: {req_id}") await send_message(client_id, {'type': 'error', 'error': 'Server not found or access denied'}) return # Check if server is a website or VPS if server['type'] not in ['website', 'vps']: await send_message(client_id, {'type': 'error', 'error': 'Domain can only be linked to Website or VPS servers'}) return # Link domain to server (this will create IngressRoute) if link_domain_to_server(domain_id, server_id): await send_message(client_id, { 'type': 'domain_linked', 'domainId': domain_id, 'serverId': server_id, 'message': f'Domain linked to server {server["name"]} with IngressRoute created' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to link domain'}) except Exception as e: logger.error(f"Error linking domain: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_verify_custom_domain(client_id, data): """Handle custom domain verification request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return domain = data.get('domain') if not domain: await send_message(client_id, {'type': 'error', 'error': 'Domain required'}) return try: # Verify domain DNS result = verify_domain_dns(domain) await send_message(client_id, { 'type': 'custom_domain_verified', 'domain': domain, 'verified': result['verified'], 'message': result['message'], 'ip_address': result.get('ip_address'), 'current_ip': result.get('current_ip') }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_create_custom_domain(client_id, data): """Handle create custom domain request with enhanced validation""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] domain = data.get('domain') server_id = data.get('serverId') if not domain: await send_message(client_id, {'type': 'error', 'error': 'Domain required'}) return # Normalize domain domain = domain.lower().strip() # Basic domain format validation if not re.match(r'^[a-z0-9.-]+\.[a-z]{2,}$', domain): await send_message(client_id, { 'type': 'error', 'error': 'Invalid domain format' }) return try: # Verify domain DNS with new restrictions verification = verify_domain_dns(domain) if verification.get('blocked'): await send_message(client_id, { 'type': 'error', 'error': verification['message'] }) return if verification.get('already_used'): await send_message(client_id, { 'type': 'error', 'error': verification['message'] }) return if not verification['verified']: await send_message(client_id, { 'type': 'error', 'error': f'Domain does not point to our server. Current IP: {verification.get("current_ip", "unknown")}' }) return # Create custom domain record domain_record = { 'user_id': user_id, 'domain': domain, 'record_type': 'A', 'ip_address': NODE_IP, 'ttl': 3600, 'server_id': server_id, 'created_at': datetime.now().isoformat() } db_domain = create_custom_domain_record(domain_record) if db_domain: await send_message(client_id, { 'type': 'custom_domain_created', 'domain': db_domain, 'message': f'Custom domain {domain} added successfully' }) else: await send_message(client_id, {'type': 'error', 'error': 'Failed to create custom domain'}) except Exception as e: logger.error(f"Error creating custom domain: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_list_custom_domains(client_id, data): """Handle list custom domains request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] try: domains = get_user_custom_domains(user_id) await send_message(client_id, { 'type': 'custom_domains_list', 'domains': domains, 'count': len(domains) }) except Exception as e: await send_message(client_id, {'type': 'error', 'error': str(e)}) async def handle_delete_custom_domain(client_id, data): """Handle delete custom domain request""" client_info = connected_clients.get(client_id) if not client_info or not client_info['authenticated']: await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return user_id = client_info['user_id'] username = client_info['username'] domain_id = data.get('domainId') if not domain_id: await send_message(client_id, {'type': 'error', 'error': 'Domain ID required'}) return try: # Get domain from database with database_connection() as conn: cursor = conn.cursor() cursor.execute(''' SELECT * FROM user_domains WHERE id = ? AND user_id = ? AND is_custom = TRUE ''', (domain_id, user_id)) domain = dict_from_row(cursor.fetchone()) if not domain: await send_message(client_id, {'type': 'error', 'error': 'Custom domain not found or access denied'}) return # If domain was linked to a server, delete IngressRoute if domain.get('server_id'): namespace = f"user-{username}".lower() # Extract subdomain for IngressRoute name domain_name = domain['domain'] subdomain = domain_name.split('.')[0] if '.' in domain_name else domain_name ingressroute_name = f"{subdomain}-ingress" # Delete IngressRoute in Kubernetes if k8s_manager.connected: try: custom_objects_api.delete_namespaced_custom_object( group="traefik.io", version="v1alpha1", namespace=namespace, plural="ingressroutes", name=ingressroute_name ) logger.info(f"Deleted IngressRoute for custom domain: {domain_name}") except ApiException as e: if e.status != 404: logger.warning(f"Error deleting IngressRoute: {e.reason}") # Delete from database cursor.execute('DELETE FROM user_domains WHERE id = ?', (domain_id,)) conn.commit() logger.info(f"✓ Deleted custom domain: {domain.get('domain')}") await send_message(client_id, { 'type': 'custom_domain_deleted', 'domainId': domain_id, 'message': 'Custom domain deleted successfully' }) except Exception as e: logger.error(f"Error deleting custom domain: {e}", exc_info=True) await send_message(client_id, {'type': 'error', 'error': str(e)}) class DateTimeEncoder(json.JSONEncoder): """Custom JSON encoder that handles datetime objects""" def default(self, obj): if isinstance(obj, (datetime, date)): return obj.isoformat() return super().default(obj) def attach_to_game_console(server_id: str, user_id: str, command: str) -> str: """Attach to game server console and send command""" try: namespace, server = get_server_access_and_namespace(server_id, user_id) if server['type'] != 'game': return "Not a game server" if not k8s_manager.connected: return f"Mock game console: {command}\n[Server] Command executed" pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return "No pods found" # Send command to game server's stdin result = k8s_manager.attach_and_send_command(namespace, pods[0], command, container_name=server_id) return result except Exception as e: logger.error(f"Error attaching to game console: {e}") return f"Error: {str(e)}" def stream_game_console_output(server_id: str, user_id: str, tail_lines: int = 50) -> str: """Get recent game console output""" try: namespace, server = get_server_access_and_namespace(server_id, user_id) if server['type'] != 'game': return "Not a game server" if not k8s_manager.connected: return "Mock game console output:\n[Server] Server is running\n[Server] Type 'help' for commands" pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return "No pods found" # Get pod logs which contain game server output logs = k8s_manager.get_pod_logs(namespace, pods[0], tail_lines, container_name=server_id) return logs except Exception as e: logger.error(f"Error streaming game console: {e}") return f"Error: {str(e)}"