"""Server access operations.""" from database_service import database_connection from datetime import datetime import json from kubernetes import client from kubernetes.client.rest import ApiException from kubernetes_service import apps_v1, k8s_manager, v1 import os import random import re from runtime_config import ( logger, ) from server_provisioning import generate_sftp_credentials, get_sftp_sidecar_container from server_records import get_server_access_and_namespace, get_server_by_id from typing import Dict, List __all__ = ['create_server_directory', 'delete_server_path', 'download_server_file', 'exec_server_command', 'expose_port', 'get_server_logs', 'get_server_metrics', 'list_server_files', 'read_server_file', 'read_server_file_base64', 'reconcile_sftp', 'remove_port', 'upload_server_file', 'validate_path_for_server', 'write_server_file'] def expose_port(server_id: str, user_id: str, internal_port: int, protocol: str = 'TCP') -> bool: """Expose a new port for a server""" try: namespace, server = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: logger.info("Kubernetes not connected, simulating port exposure") # Update database with new port if server: ports = server['ports'] external_port = random.randint(30000, 32768) ports.append({ 'internal': internal_port, 'external': external_port, 'protocol': protocol, 'name': f'port-{internal_port}-{protocol}'.lower() }) with database_connection() as conn: cursor = conn.cursor() cursor.execute( 'UPDATE servers SET ports = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (json.dumps(ports), server_id) ) conn.commit() return True service_name = f"{server_id}-service" # Get existing service try: service = v1.read_namespaced_service(name=service_name, namespace=namespace) except ApiException as e: if e.status == 404: # Service doesn't exist, create it external_port = random.randint(30000, 32768) service_ports = [ client.V1ServicePort( name=f"port-{internal_port}-{protocol}".lower(), port=internal_port, target_port=internal_port, node_port=external_port, protocol=protocol ) ] service = client.V1Service( metadata=client.V1ObjectMeta( name=service_name, namespace=namespace, labels={'app': server_id} ), spec=client.V1ServiceSpec( selector={'app': server_id}, type='NodePort', ports=service_ports ) ) created_service = v1.create_namespaced_service(namespace=namespace, body=service) # Get the actual assigned node port actual_external_port = created_service.spec.ports[0].node_port # Update database server = get_server_by_id(server_id) if server: ports = [{ 'internal': internal_port, 'external': actual_external_port, 'protocol': protocol, 'name': f'port-{internal_port}-{protocol}'.lower() }] with database_connection() as conn: cursor = conn.cursor() cursor.execute( 'UPDATE servers SET ports = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (json.dumps(ports), server_id) ) conn.commit() return True else: raise e # Check if port already exists for port in service.spec.ports: if port.port == internal_port and port.protocol == protocol: return True # Port already exposed # Add new port external_port = random.randint(30000, 32768) new_port = client.V1ServicePort( name=f"port-{internal_port}-{protocol}".lower(), port=internal_port, target_port=internal_port, node_port=external_port, protocol=protocol ) service.spec.ports.append(new_port) updated_service = v1.patch_namespaced_service(name=service_name, namespace=namespace, body=service) # Get the actual assigned node port actual_external_port = None for port in updated_service.spec.ports: if port.port == internal_port and port.protocol == protocol: actual_external_port = port.node_port break # Update server ports in database server = get_server_by_id(server_id) if server: ports = server['ports'] ports.append({ 'internal': internal_port, 'external': actual_external_port or external_port, 'protocol': protocol, 'name': f'port-{internal_port}-{protocol}'.lower() }) with database_connection() as conn: cursor = conn.cursor() cursor.execute( 'UPDATE servers SET ports = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (json.dumps(ports), server_id) ) conn.commit() return True except Exception as e: logger.error(f"Error exposing port: {e}") raise Exception(f'Failed to expose port: {str(e)}') def remove_port(server_id: str, user_id: str, internal_port: int, protocol: str = 'TCP') -> bool: """Remove an exposed port from a server""" try: namespace, server = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: logger.info("Kubernetes not connected, simulating port removal") if server: ports = server['ports'] new_ports = [p for p in ports if not (p['internal'] == internal_port and p['protocol'] == protocol)] with database_connection() as conn: cursor = conn.cursor() cursor.execute( 'UPDATE servers SET ports = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (json.dumps(new_ports), server_id) ) conn.commit() return True service_name = f"{server_id}-service" # Get existing service try: service = v1.read_namespaced_service(name=service_name, namespace=namespace) except ApiException as e: if e.status == 404: return True # Service doesn't exist, nothing to remove else: raise e # Remove port from service - FIXED: Check both port number and protocol service.spec.ports = [p for p in service.spec.ports if not (p.port == internal_port and p.protocol == protocol)] # If no ports left, delete the service if not service.spec.ports: v1.delete_namespaced_service(name=service_name, namespace=namespace) else: # Update the service with remaining ports - Use replace instead of patch for reliable array updates # Clear managed_fields to avoid conflicts with server-side apply if hasattr(service.metadata, 'managed_fields') and service.metadata.managed_fields: service.metadata.managed_fields = None v1.replace_namespaced_service(name=service_name, namespace=namespace, body=service) # Update server ports in database server = get_server_by_id(server_id) if server: ports = [p for p in server['ports'] if not (p['internal'] == internal_port and p['protocol'] == protocol)] with database_connection() as conn: cursor = conn.cursor() cursor.execute( 'UPDATE servers SET ports = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (json.dumps(ports), server_id) ) conn.commit() logger.info(f"✅ Successfully removed port {internal_port}/{protocol} from server {server_id}") return True except Exception as e: logger.error(f"Error removing port: {e}") raise Exception(f'Failed to remove port: {str(e)}') def get_server_logs(server_id: str, user_id: str) -> str: try: namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: return f"Mock logs for {server_id}" pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return f"No pods found" logs = k8s_manager.get_pod_logs(namespace, pods[0], container_name=server_id) return logs except Exception as e: return f"Error: {str(e)}" def reconcile_sftp(server_id: str, user_id: str) -> bool: """Ensure SFTP sidecar and port are correctly configured for a server""" try: logger.info(f"Checking SFTP reconciliation for server {server_id}") namespace, server = get_server_access_and_namespace(server_id, user_id) if server['type'] == 'vps' or not k8s_manager.connected: logger.info(f"Skipping SFTP for {server_id}: type={server['type']}, k8s_connected={k8s_manager.connected}") return True # 1. Ensure credentials exist u = server.get('sftp_username') p = server.get('sftp_password') port = server.get('sftp_port') if not u or not p or not port: logger.info(f"Generating missing SFTP credentials for {server_id}") 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() server['sftp_username'] = u server['sftp_password'] = p server['sftp_port'] = port # 2. Ensure sidecar in Deployment try: deployment = apps_v1.read_namespaced_deployment(name=server_id, namespace=namespace) containers = deployment.spec.template.spec.containers sftp_container_index = next((i for i, c in enumerate(containers) if c.name == 'sftp'), -1) # Determine mount path based on image mount_path = '/data' if server['image'] == 'nginx': mount_path = '/usr/share/nginx/html' elif server['image'] in ['wordpress', 'wordpress-sqlite', 'node']: mount_path = '/var/www/html' elif server['image'] == 'unturned': mount_path = '/home/steam/Unturned' elif server['image'] == 'factorio': mount_path = '/factorio' elif server['image'] == 'hytale': mount_path = '/home/container' elif server['image'] == 'zomboid': mount_path = '/home/steam/Zomboid' sftp_container = get_sftp_sidecar_container(server_id, u, p, mount_path) container_spec = client.ApiClient().sanitize_for_serialization(sftp_container) if sftp_container_index == -1: logger.info(f"Adding SFTP sidecar to deployment {server_id}") patch = [ { "op": "add", "path": "/spec/template/spec/containers/-", "value": container_spec } ] apps_v1.patch_namespaced_deployment( name=server_id, namespace=namespace, body=patch, _content_type='application/json-patch+json' ) logger.info(f"✓ Successfully added SFTP sidecar to deployment {server_id}") else: # Update existing sidecar if needed (check if env, image or security_context differ) existing_container = containers[sftp_container_index] needs_update = False # Check SFTP_USERS env existing_sftp_users = next((env.value for env in existing_container.env if env.name == 'SFTP_USERS'), None) expected_sftp_users = f"{u}:{p}:1000:1000:data" # Check image existing_image = existing_container.image if existing_sftp_users != expected_sftp_users or existing_image != "atmoz/sftp:alpine": logger.info(f"Updating SFTP configuration for {server_id} (image or credentials changed)") needs_update = True if needs_update: patch = [ { "op": "replace", "path": f"/spec/template/spec/containers/{sftp_container_index}", "value": container_spec } ] apps_v1.patch_namespaced_deployment( name=server_id, namespace=namespace, body=patch, _content_type='application/json-patch+json' ) logger.info(f"✓ Successfully updated SFTP sidecar for deployment {server_id}") else: logger.info(f"SFTP sidecar already exists and is up to date for {server_id}") except Exception as e: logger.error(f"Error reconciling SFTP sidecar for {server_id}: {e}") # 3. Ensure port in Service try: service_name = f"{server_id}-service" service = v1.read_namespaced_service(name=service_name, namespace=namespace) ports = service.spec.ports sftp_port_exists = any(p.name == 'sftp' for p in ports) if not sftp_port_exists: logger.info(f"Adding SFTP port {port} to service {service_name}") sftp_service_port = client.V1ServicePort( name="sftp", port=22, target_port=22, node_port=int(port), protocol="TCP" ) port_spec = client.ApiClient().sanitize_for_serialization(sftp_service_port) patch = [ { "op": "add", "path": "/spec/ports/-", "value": port_spec } ] v1.patch_namespaced_service( name=service_name, namespace=namespace, body=patch, _content_type='application/json-patch+json' ) logger.info(f"✓ Successfully added SFTP port to service {service_name}") else: logger.info(f"SFTP port already exists for {service_name}") except Exception as e: logger.error(f"Error reconciling SFTP port for {server_id}: {e}") return True except Exception as e: logger.error(f"Critical error in reconcile_sftp for {server_id}: {e}") return False def get_server_metrics(server_id: str, user_id: str) -> Dict: try: namespace, server = get_server_access_and_namespace(server_id, user_id) server_ram_gb = 1 server_cpu_cores = 1 server_ram_gb = int(server['ram'].replace('GB', '')) if 'GB' in server['ram'] else 1 # Parse CPU cores from server config cpu_str = server.get('cpu', '1') if 'vCPU' in cpu_str: server_cpu_cores = int(cpu_str.split()[0]) if cpu_str.split()[0].isdigit() else 1 else: match = re.search(r'(\d+)', cpu_str) if match: server_cpu_cores = int(match.group(1)) pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: # Use fallback metrics when no pods found cpu_millicores = random.randint(50, 300) memory_mebibytes = random.randint(128, 1024) max_cpu_millicores = server_cpu_cores * 1000 cpu_percentage = min((cpu_millicores / max_cpu_millicores) * 100, 100) if max_cpu_millicores > 0 else 0 memory_percentage = min((memory_mebibytes / (server_ram_gb * 1024)) * 100, 100) return { 'cpu': f'{cpu_millicores}m', 'memory': f'{memory_mebibytes}Mi', 'cpu_percentage': round(cpu_percentage, 1), 'memory_percentage': round(memory_percentage, 1), 'timestamp': datetime.now().isoformat() } metrics = k8s_manager.get_pod_metrics(namespace, pods[0], server_ram_gb, server_cpu_cores) return metrics except Exception as e: # Use fallback metrics on error cpu_millicores = random.randint(50, 300) memory_mebibytes = random.randint(128, 1024) server_cpu_cores = 1 # Default fallback server_ram_gb = 1 # Default fallback max_cpu_millicores = server_cpu_cores * 1000 cpu_percentage = min((cpu_millicores / max_cpu_millicores) * 100, 100) if max_cpu_millicores > 0 else 0 memory_percentage = min((memory_mebibytes / (server_ram_gb * 1024)) * 100, 100) return { 'cpu': f'{cpu_millicores}m', 'memory': f'{memory_mebibytes}Mi', 'cpu_percentage': round(cpu_percentage, 1), 'memory_percentage': round(memory_percentage, 1), 'error': str(e), 'timestamp': datetime.now().isoformat() } def validate_path_for_server(server_id: str, path: str) -> str: """Validate and normalize path for server, restricting servers to their data directories""" server = get_server_by_id(server_id) if not server: return path server_type = server.get('type') image = server.get('image') # Image to base path mapping base_paths = { # Game servers 'minecraft': '/data', 'rust': '/data', 'unturned': '/home/steam/Unturned', 'factorio': '/factorio', 'hytale': '/home/container', 'zomboid': '/home/steam/Zomboid', # Website servers 'nginx': '/usr/share/nginx/html', 'wordpress': '/var/www/html', 'node': '/var/www/html' } # Determine base path if server_type == 'game': base_path = base_paths.get(image, '/data') elif server_type == 'website': base_path = base_paths.get(image, '/var/www/html') else: # For VPS or other types, allow full path or define specific root if needed return path # If path is relative, make it absolute starting from base_path if not path.startswith('/'): path = os.path.join(base_path, path) elif not path.startswith(base_path): # If it's absolute but doesn't start with base_path, prefix it with base_path # e.g. /etc/config -> /data/etc/config path = os.path.join(base_path, path.lstrip('/')) # Ensure path doesn't go above base path normalized = os.path.normpath(path) if not normalized.startswith(base_path): return base_path return normalized def list_server_files(server_id: str, user_id: str, path: str = "/") -> List[Dict]: """List files in server with improved error handling""" try: # Validate path for Minecraft servers path = validate_path_for_server(server_id, path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: # Return mock data for development return [ {"name": "eula.txt", "type": "file", "size": 1024, "modified": "2024-01-01T00:00:00Z"}, {"name": "server.properties", "type": "file", "size": 2048, "modified": "2024-01-01T00:00:00Z"}, {"name": "logs", "type": "directory", "size": 4096, "modified": "2024-01-01T00:00:00Z"}, {"name": "world", "type": "directory", "size": 4096, "modified": "2024-01-01T00:00:00Z"} ] pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return [] return k8s_manager.list_files(namespace, pods[0], path, container_name=server_id) except Exception as e: logger.error(f"Error listing server files: {e}") return [] def read_server_file(server_id: str, user_id: str, file_path: str) -> str: try: # Validate path for Minecraft servers file_path = validate_path_for_server(server_id, file_path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: return "" pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return "" return k8s_manager.read_file(namespace, pods[0], file_path, container_name=server_id) except Exception: return "" def read_server_file_base64(server_id: str, user_id: str, file_path: str) -> str: """Read server file as base64 encoded string""" try: # Validate path for Minecraft servers file_path = validate_path_for_server(server_id, file_path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: return "" pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return "" return k8s_manager.read_file_base64(namespace, pods[0], file_path, container_name=server_id) except Exception: return "" def write_server_file(server_id: str, user_id: str, file_path: str, content: str) -> bool: try: # Validate path for Minecraft servers file_path = validate_path_for_server(server_id, file_path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: return True pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return False return k8s_manager.write_file(namespace, pods[0], file_path, content, container_name=server_id) except Exception: return False def create_server_directory(server_id: str, user_id: str, dir_path: str) -> bool: """Create directory in server container""" try: # Validate path for Minecraft servers dir_path = validate_path_for_server(server_id, dir_path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: return True pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return False return k8s_manager.create_directory(namespace, pods[0], dir_path, container_name=server_id) except Exception: return False def delete_server_path(server_id: str, user_id: str, path: str) -> bool: """Delete file or directory in server container""" try: # Validate path for Minecraft servers path = validate_path_for_server(server_id, path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: return True pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return False return k8s_manager.delete_path(namespace, pods[0], path, container_name=server_id) except Exception: return False def upload_server_file(server_id: str, user_id: str, file_path: str, file_content: str, is_base64: bool = True) -> bool: """Upload file to server container (now uses binary upload for all files)""" try: # Validate path for Minecraft servers file_path = validate_path_for_server(server_id, file_path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: return True pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return False # Use write_file_binary for all files to support binary files return k8s_manager.write_file_binary(namespace, pods[0], file_path, file_content, container_name=server_id) except Exception as e: logger.error(f"Error in upload_server_file: {e}") return False def download_server_file(server_id: str, user_id: str, file_path: str) -> str: """Download file from server container""" try: # Validate path for Minecraft servers file_path = validate_path_for_server(server_id, file_path) namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: logger.warning("K8s manager not connected, cannot download file") return "" pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: logger.error(f"No pods found for server {server_id}") return "" return k8s_manager.download_file(namespace, pods[0], file_path, container_name=server_id) except Exception as e: logger.error(f"Error downloading file {file_path} from server {server_id}: {e}", exc_info=True) return "" def exec_server_command(server_id: str, user_id: str, command: str) -> str: """Execute command in server container""" try: namespace, _ = get_server_access_and_namespace(server_id, user_id) if not k8s_manager.connected: return f"$ {command}\nMock command execution completed" pods = k8s_manager.get_pods_for_server(namespace, server_id) if not pods: return "" return k8s_manager.exec_command(namespace, pods[0], ['sh', '-c', command], container_name=server_id) except Exception: return ""