import os import select import time import uuid import subprocess import threading from flask import Blueprint, request, current_app, session from flask_socketio import Namespace, join_room, leave_room, emit, SocketIO, rooms, disconnect from threading import Thread import sys from ..extensions import socketio import pty import termios import struct import fcntl import psutil TERM_INIT_CONFIG = { # instead of local server runnning this web terminal service # "domain" is the target that you want to access through local server (with this web terminal) # and before doing so - make sure you have username and port (on the "domain") to implement remote access 'domain': '127.0.0.1', # or ip address like 192.168.10.11 'client_path': { 'telnet': '/usr/bin/telnet', # confirmed location of your client binary (with cmd like 'which telnet') 'ssh': '/usr/bin/ssh' } } terminal_bp = Blueprint('terminal', __name__, url_prefix='/vsc-like') def set_winsize(fd, row, col, xpix=0, ypix=0): try: winsize = struct.pack("HHHH", row, col, xpix, ypix) fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize) except (OSError, IOError): # File descriptor closed or invalid, do nothing pass def read_and_forward_pty_output(fd=None, pid=None, room_id=None, namespace=None): """ read data on pty master from the pty slave, and emit to the web terminal visitor """ max_read_bytes = 1024 * 20 # 初始超时时间设置为较短值，确保响应迅速 timeout = 0.05 # 最大超时时间，避免太频繁的检查 max_timeout = 0.5 # 成功读取数据后的重置超时时间 reset_timeout = 0.05 try: while True: # 等待一段时间，减少CPU占用 socketio.sleep(timeout) # 使用指数退避算法调整超时时间，但不超过最大值 timeout = min(timeout * 1.5, max_timeout) # 检查子进程状态 try: child_process = psutil.Process(pid) except psutil.NoSuchProcess: # 进程已终止，清理僵尸进程 try: os.waitpid(pid, os.WNOHANG) except Exception: pass return # 检查进程状态，如果不是运行或睡眠状态，则退出 if child_process.status() not in ('running', 'sleeping'): try: # 等待进程终止 child_process.wait(timeout=1) except Exception: try: os.waitpid(pid, os.WNOHANG) except Exception: pass return # 如果文件描述符有效，尝试读取数据 if fd: try: # 使用非阻塞select检查是否有数据可读 (data_ready, _, _) = select.select([fd], [], [], 0) if data_ready: # 有数据可读，重置超时时间 timeout = reset_timeout try: # 读取数据 output = os.read(fd, max_read_bytes).decode() except (OSError, IOError, EOFError): # 文件描述符已关闭或其他IO错误，优雅退出 return except UnicodeDecodeError as err: # 处理编码错误 output = f""" ***AQUI WEB TERM ERR*** Unicode decode error: {err} *********************** """ # 发送数据到客户端 try: namespace.emit("pty_output", {"output": output}, room=room_id) except Exception: # 如果发送失败，客户端可能已断开连接，退出 return except (OSError, IOError): # 文件描述符无效，优雅退出 return except Exception as e: # 捕获任何其他未预期的异常，防止服务器崩溃 current_app.logger.error(f"Unexpected error in read_and_forward_pty_output: {e}") finally: # 清理文件描述符 if fd: try: os.close(fd) except Exception: pass class VSCLikeNameSpace(Namespace): def create_terminal(self): """创建新的终端并更新session配置""" terminal_config = session.get('terminal_config', {}) # 如果没有房间ID，创建一个新的 if not terminal_config.get('room_id'): session_id = str(uuid.uuid4()) room_id = f"terminal_{session_id}" join_room(room_id) terminal_config.update({ **TERM_INIT_CONFIG.copy(), 'session_id': session_id, 'room_id': room_id }) else: room_id = terminal_config['room_id'] current_app.logger.debug(f"Creating new terminal for room: {room_id}") # create child process attached to a pty we can read from and write to (child_pid, fd) = pty.fork() lesson_path = session.get('path') if child_pid == 0: # this is the child process fork. # anything printed here will show up in the pty, including the output # of this subprocess try: # 获取终端配置 term_type = terminal_config.get('term_type') if not term_type: print("Terminal type not specified, exit") os._exit(1) path = TERM_INIT_CONFIG.get('client_path', {}).get(term_type, None) if not path or not os.path.exists(path): print(f"Can't locate {term_type} binary at {path}, exit") os._exit(1) # 获取连接参数 username = terminal_config.get('username') domain = terminal_config.get('domain', TERM_INIT_CONFIG['domain']) port = terminal_config.get('port') if not username or not domain: print("Missing required connection parameters, exit") os._exit(1) if term_type == 'telnet': # 使用telnet连接 if not port: print("Port not specified for telnet, exit") os._exit(1) os.execl(path, 'telnet', '-l', username, domain, str(port)) elif term_type == 'ssh': # 使用ssh连接 ssh_port = port if port else 22 # 默认端口22 os.execl(path, 'ssh', '-p', str(ssh_port), f'{username}@{domain}') else: print(f"Wrong term type {term_type}, exit") os._exit(1) except Exception as e: print(f"Error in child process: {e}") os._exit(1) else: # 更新会话配置，保存到session中 terminal_config['fd'] = fd terminal_config['child_pid'] = child_pid session['terminal_config'] = terminal_config session.modified = True # 设置初始窗口大小 set_winsize(fd, 50, 50) # 记录日志 current_app.logger.debug(f"Child process created: {child_pid} for room: {room_id}") # 启动后台任务读取pty输出，传入正确的room_id socketio.start_background_task(read_and_forward_pty_output, fd, child_pid, room_id, self) current_app.logger.debug("Background task running") return terminal_config def on_connect(self): """new client connected""" # 为每个客户端生成唯一的会话ID和房间ID session_id = str(uuid.uuid4()) room_id = f"terminal_{session_id}" # 加入房间 join_room(room_id) # 确保terminal_config存在于session中，每个客户端有独立配置 terminal_config = { **TERM_INIT_CONFIG.copy(), 'session_id': session_id, 'room_id': room_id } session['terminal_config'] = terminal_config session.modified = True current_app.logger.debug(f"Client connected, session_id: {session_id}, room_id: {room_id}") # 创建终端 self.create_terminal() current_app.logger.debug("background task running") def on_pty_input(self, data): """write to the child pty, which now is the ssh process from this machine to the 'domain' configured """ # 输入验证 if not isinstance(data, dict): current_app.logger.error("Invalid input format: expected dictionary") return input_data = data.get("input") if not isinstance(input_data, str): current_app.logger.error("Invalid input data: expected string") return # 限制输入长度，防止缓冲区溢出 if len(input_data) > 1024: input_data = input_data[:1024] current_app.logger.warning("Input truncated due to length") try: terminal_config = session.get('terminal_config', {}) child_pid = terminal_config.get('child_pid') # 如果没有子进程，重新创建终端 if not child_pid: current_app.logger.debug("No child process found, creating new terminal") terminal_config = self.create_terminal() child_pid = terminal_config.get('child_pid') if not child_pid: current_app.logger.error("Failed to create new terminal") return # 检查子进程是否存在且运行中 try: child_process = psutil.Process(child_pid) if child_process.status() not in ('running', 'sleeping'): current_app.logger.debug("Child process not running, creating new terminal") terminal_config = self.create_terminal() child_pid = terminal_config.get('child_pid') if not child_pid: current_app.logger.error("Failed to create new terminal") return except psutil.NoSuchProcess: current_app.logger.debug("Child process not found, creating new terminal") terminal_config = self.create_terminal() child_pid = terminal_config.get('child_pid') if not child_pid: current_app.logger.error("Failed to create new terminal") return # 获取文件描述符并写入数据 fd = terminal_config.get('fd') if fd: try: os.write(fd, input_data.encode()) except (OSError, IOError) as e: # File descriptor closed or invalid, create new terminal current_app.logger.debug(f"Error writing to file descriptor: {e}, creating new terminal") terminal_config = self.create_terminal() except Exception as e: current_app.logger.error(f"Error in pty_input handling: {e}") # 发生异常时尝试重新创建终端 try: self.create_terminal() except Exception as create_err: current_app.logger.error(f"Failed to recreate terminal: {create_err}") def on_resize(self, data): # 输入验证 if not isinstance(data, dict): current_app.logger.error("Invalid resize data format: expected dictionary") return rows = data.get("rows") cols = data.get("cols") # 验证行和列的值是否为正整数 if not (isinstance(rows, int) and isinstance(cols, int)): current_app.logger.error("Invalid resize dimensions: expected integers") return # 限制窗口大小范围，防止异常值 if rows <= 0 or rows > 1000 or cols <= 0 or cols > 1000: current_app.logger.error(f"Invalid resize dimensions: {rows}x{cols} (out of range)") return try: terminal_config = session.get('terminal_config', {}) child_pid = terminal_config.get('child_pid') # 如果没有子进程，重新创建终端 if not child_pid: current_app.logger.debug("No child process found for resize, creating new terminal") terminal_config = self.create_terminal() child_pid = terminal_config.get('child_pid') if not child_pid: current_app.logger.error("Failed to create new terminal for resize") return # 检查子进程是否存在且运行中 try: child_process = psutil.Process(child_pid) if child_process.status() not in ('running', 'sleeping'): current_app.logger.debug("Child process not running for resize, creating new terminal") terminal_config = self.create_terminal() child_pid = terminal_config.get('child_pid') if not child_pid: current_app.logger.error("Failed to create new terminal for resize") return except psutil.NoSuchProcess: current_app.logger.debug("Child process not found for resize, creating new terminal") terminal_config = self.create_terminal() child_pid = terminal_config.get('child_pid') if not child_pid: current_app.logger.error("Failed to create new terminal for resize") return # 获取文件描述符并调整窗口大小 fd = terminal_config.get('fd') if fd: try: os.fstat(fd) set_winsize(fd, rows, cols) except (OSError, IOError) as e: # 文件描述符无效，创建新终端 current_app.logger.debug(f"Error resizing terminal: {e}, creating new terminal") terminal_config = self.create_terminal() except Exception as e: current_app.logger.error(f"Error in resize handling: {e}") # 发生异常时尝试重新创建终端 try: self.create_terminal() except Exception as create_err: current_app.logger.error(f"Failed to recreate terminal for resize: {create_err}") def on_disconnect(self): terminal_config = session.get('terminal_config', {}) child_pid = terminal_config.get('child_pid') fd = terminal_config.get('fd') room_id = terminal_config.get('room_id') session_id = terminal_config.get('session_id') current_app.logger.debug(f"Client disconnecting, session_id: {session_id}, room_id: {room_id}") # 退出房间 if room_id: leave_room(room_id) current_app.logger.debug(f"Client left room: {room_id}") # 先关闭文件描述符，避免其他操作尝试使用无效的文件描述符 if fd: try: os.close(fd) current_app.logger.debug(f"Closed file descriptor: {fd}") except Exception as e: current_app.logger.debug(f"Error closing file descriptor: {e}") if child_pid: try: child_process = psutil.Process(child_pid) if child_process.status() in ('running', 'sleeping'): # if visitor just close the browser tab then left alone the pty here # it should be terminated by the parent process after child_process.terminate() # Wait for the process to terminate and collect its exit status child_process.wait(timeout=2) current_app.logger.debug(f'User left pty alone, terminated and waited: {child_pid}') except psutil.NoSuchProcess: # Process already terminated, try to wait anyway to clean up any zombie try: os.waitpid(child_pid, os.WNOHANG) current_app.logger.debug(f'Cleaned up zombie process: {child_pid}') except Exception as e: current_app.logger.debug(f"Error cleaning up zombie process: {e}") except psutil.TimeoutExpired: # If process didn't terminate in time, kill it forcefully try: child_process.kill() child_process.wait(timeout=1) current_app.logger.debug(f'Forcefully killed process: {child_pid}') except Exception as e: current_app.logger.debug(f"Error forcefully killing process: {e}") except Exception as err: current_app.logger.error(f'Error terminating process: {err}') # Clear session config for this client session.pop('terminal_config', None) session.modified = True current_app.logger.debug(f"Client disconnected, session_id: {session_id}, room_id: {room_id}") current_app.logger.debug('Client disconnected')