""" Heady Remote Access - FastAPI implementation Secure guacamole-based remote access for VPN infrastructure """ import asyncio import logging import threading import traceback from typing import Optional from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, HTTPException, status from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import os from guacamole.client import GuacamoleClient from guacamole.instruction import GuacamoleInstruction from auth import get_current_user, User from session_manager import SessionManager, RemoteSession # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # FastAPI app app = FastAPI( title="Heady Remote Access", description="Secure remote access for awesome VPN management", version="1.0.0" ) # CORS middleware for development app.add_middleware( CORSMiddleware, allow_origins=["*"], # Configure appropriately for production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Session manager session_manager = SessionManager() # Configuration GUACD_HOST = os.getenv("GUACD_HOST", "guacd") GUACD_PORT = int(os.getenv("GUACD_PORT", "4822")) HEADPLANE_API_URL = os.getenv("HEADPLANE_API_URL", "http://headplane:3000") class TerminalSession: """ Terminal session handler - adapted from your Django consumer """ def __init__(self, websocket: WebSocket, user: User, node_name: str, protocol: str = "ssh"): self.websocket = websocket self.user = user self.node_name = node_name self.protocol = protocol self.client: Optional[GuacamoleClient] = None self.stopping = False self.access_level = "full" # TODO: Implement based on OIDC roles async def connect(self): """Initialize guacamole connection - based on your Django logic""" try: # Create session record self.session = await session_manager.create_session( user=self.user, node_name=self.node_name, protocol=self.protocol ) # Build guacd connection parameters - multi-protocol support guacd_params = self._build_protocol_params() # Add recording if configured recording_config = await session_manager.get_recording_config(self.session.id) if recording_config: guacd_params.update(recording_config) # Connect to guacd self.client = GuacamoleClient(GUACD_HOST, GUACD_PORT) logger.info(f"Connecting to {self.node_name} via {self.protocol} for user {self.user.email}") self.client.handshake(**guacd_params) if self.client.connected: # Accept WebSocket and start data flow await self.websocket.accept(subprotocol="guacamole") # Start receiving from guacd in background task await asyncio.create_task(self._start_guacd_receiver()) return True else: await self._send_error("Failed to connect to target system", 771) return False except (ConnectionRefusedError, OSError) as e: logger.error(f"Failed connecting to guacd @ {GUACD_HOST}:{GUACD_PORT}: {e}") await self._send_error("Failed connection to guacd", 771) return False except Exception as e: logger.error(f"Error in connect: {e}\\n{traceback.format_exc()}") await self._send_error("System error. Please close and try again.", 100) return False async def _start_guacd_receiver(self): """Start background task to receive data from guacd""" def receive_from_guacd(): """Receive data from guacd - adapted from your open() method""" while not self.stopping: try: content = self.client.receive() if content: # Skip 0x0 layer size workaround from your implementation if content == "4.size,1.1,1.0,1.0;": continue # Send to WebSocket via async call asyncio.create_task(self._send_to_websocket(content)) except OSError: # Client disconnected break # Run in thread to avoid blocking async loop thread = threading.Thread(target=receive_from_guacd) thread.daemon = True thread.start() async def _send_to_websocket(self, content: str): """Send content to WebSocket safely""" try: await self.websocket.send_text(content) except Exception: # WebSocket closed self.stopping = True async def receive_data(self, data: str): """Handle data from WebSocket - adapted from your receive() method""" # Access control from your implementation if self.access_level != "full" and (data.startswith("5.mouse") or data.startswith("3.key")): return # Skip ping messages if data.startswith("0.,"): return # Send to guacd if self.client and data: self.client.send(data) async def disconnect(self): """Clean up connection - adapted from your disconnect() method""" self.stopping = True if self.client: self.client.close() # Update session duration if hasattr(self, 'session'): await session_manager.close_session(self.session.id) logger.info(f"Disconnected from {self.node_name} for user {self.user.email}") def _build_protocol_params(self) -> dict: """ Build guacd connection parameters for different protocols Supports SSH, RDP, VNC, Telnet, Kubernetes - BLAM! 💥 """ base_params = { "protocol": self.protocol, "hostname": self.node_name, "username": self.user.preferred_username, "width": 1024, "height": 768, "dpi": 96 } if self.protocol == "ssh": return { **base_params, "port": 22, "font-size": 12, "color-scheme": "gray-black", "enable-sftp": "true", # File transfer support "sftp-root-directory": "/home/" + self.user.preferred_username } elif self.protocol == "rdp": return { **base_params, "port": 3389, "security": "any", "ignore-cert": "true", "enable-drive": "true", # Drive mapping "drive-path": "/drive", "console": "true", "width": 1920, # Better resolution for RDP "height": 1080, "color-depth": 24 } elif self.protocol == "vnc": return { **base_params, "port": 5901, "password": "", # VNC password if required "enable-sftp": "false", "color-depth": 24, "cursor": "remote" } elif self.protocol == "telnet": return { **base_params, "port": 23, "font-size": 12, "color-scheme": "green-black", # Classic terminal look "backspace": "127" } elif self.protocol == "kubernetes": # Special handling for kubectl exec sessions return { **base_params, "port": 22, # Usually SSH to k8s node "font-size": 12, "color-scheme": "blue-black", # Distinct k8s look "command": f"kubectl exec -it {self.node_name} -- /bin/bash", "enable-sftp": "false" # No file transfer for k8s } else: # Default to SSH return { **base_params, "port": 22 } async def _send_error(self, error_text: str, error_code: int): """Send error message to client""" await self.websocket.accept() error_instruction = GuacamoleInstruction("error", error_text, error_code) await self.websocket.send_text(error_instruction.encode()) @app.post("/api/terminal/token") async def create_terminal_token( request: dict, user: User = Depends(get_current_user) ): """ Create encrypted connection token for guacamole-lite Provides secure, server-side credential management """ from token_encryption import create_connection_token from auth import get_terminal_permissions, check_terminal_access node_id = request.get("node_id") protocol = request.get("protocol", "ssh") width = request.get("width", 1024) height = request.get("height", 768) readonly = request.get("readonly", False) # Validate access permissions if not await check_terminal_access(user, node_id, protocol): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Access denied: {protocol.upper()} not permitted for your role" ) # Get user permissions for additional settings permissions = get_terminal_permissions(user.groups) # Build additional parameters based on permissions and protocol additional_params = { "width": str(width), "height": str(height), "readonly": "true" if readonly else "false" } # Add recording if required for this user role if permissions.get("session_recording", False): additional_params["recording_path"] = f"/recordings/{user.id}" # Add file transfer settings for SSH if protocol == "ssh" and permissions.get("file_transfer", False): additional_params.update({ "enable-sftp": "true", "sftp-root-directory": f"/home/{user.preferred_username}" }) try: # Create encrypted token token = create_connection_token( protocol=protocol, hostname=node_id, username=user.preferred_username, additional_params=additional_params ) logger.info(f"Created connection token for {user.email} -> {node_id} ({protocol})") return { "token": token, "websocket_url": f"ws://localhost:8000/guacamole-websocket-tunnel", "expires_in": 3600 # Token valid for 1 hour } except Exception as e: logger.error(f"Failed to create token: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create connection token" ) @app.websocket("/terminal/{node_name}") async def terminal_endpoint( websocket: WebSocket, node_name: str, protocol: str = "ssh", user: User = Depends(get_current_user) ): """ Multi-protocol WebSocket endpoint - SSH, RDP, VNC, Telnet, Kubernetes Protocol support based on user role permissions """ session = TerminalSession(websocket, user, node_name, protocol) try: # Connect and validate access connected = await session.connect() if not connected: return # Handle WebSocket messages while True: try: data = await websocket.receive_text() await session.receive_data(data) except WebSocketDisconnect: break except Exception as e: logger.error(f"Terminal session error: {e}\\n{traceback.format_exc()}") finally: await session.disconnect() @app.get("/health") async def health_check(): """Health check endpoint""" return {"status": "awesome", "service": "heady-remote-access"} @app.get("/sessions") async def list_sessions(user: User = Depends(get_current_user)): """List active sessions for user""" sessions = await session_manager.get_user_sessions(user.id) return {"sessions": sessions} @app.delete("/sessions/{session_id}") async def terminate_session( session_id: str, user: User = Depends(get_current_user) ): """Terminate a specific session""" success = await session_manager.terminate_session(session_id, user.id) if not success: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Session not found or access denied" ) return {"message": "Session terminated"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)