package middleware import ( "blazing/common/utils" "blazing/cool" "blazing/modules/config/model" config "blazing/modules/config/service" "bufio" "context" "crypto/rand" "encoding/hex" "encoding/json" "fmt" "io" "net" "strings" "sync/atomic" "time" "github.com/gogf/gf/v2/os/glog" "github.com/gogf/gf/v2/util/grand" "github.com/lxzan/gws" "golang.org/x/crypto/ssh" ) const ( PingInterval = 10 * time.Second defaultWorkDir = "$HOME" // 全环境兼容 randomStrLength = 4 // 缩短随机串长度 cmdTimeout = 180 * time.Second // 延长超时(适配Screen安装+下载) ) // SSHConfig SSH连接配置 type SSHConfig struct { ServerIP string `json:"server_ip"` Port string `json:"port"` Username string `json:"username"` Password string `json:"password"` } // WebSSHMessage 消息结构 type WebSSHMessage struct { Type string `json:"type"` Payload string `json:"payload"` } // TerminalSession 会话结构体(优化通道) type TerminalSession struct { WebSocket *gws.Conn SSHClient *ssh.Client SSHSession *ssh.Session Stdin io.WriteCloser Stdout io.Reader Ready bool outputBuf chan string // 增大通道缓存 closed atomic.Bool } type ServerHandler struct { gws.BuiltinEventHandler session *TerminalSession model.ServerList isinstall uint32 target net.Conn } // 生成随机文件名 func (s *ServerHandler) generateRandomFileName() string { randBytes := make([]byte, randomStrLength) _, err := rand.Read(randBytes) if err != nil { return fmt.Sprintf("logic_%d", time.Now().UnixNano()) } randStr := hex.EncodeToString(randBytes) timestamp := time.Now().Format("20060102150405") return fmt.Sprintf("logic_%s_%s", timestamp, randStr) } // 执行脚本命令 func (s *ServerHandler) executeScript(scriptContent, scriptName string) (string, error) { if s.session == nil || s.session.closed.Load() || !s.session.Ready { return "", fmt.Errorf("session not ready") } // 生成临时脚本路径 scriptPath := fmt.Sprintf("/tmp/%s.sh", scriptName) // 直接将脚本内容写入文件 writeScriptCmd := fmt.Sprintf("cat > '%s' << 'DEPLOYMENT_SCRIPT_END'\n%s\nDEPLOYMENT_SCRIPT_END\n", scriptPath, scriptContent) _, err := s.session.Stdin.Write([]byte(writeScriptCmd)) if err != nil { return "", err } // 等待一会儿确保脚本写入完成 time.Sleep(time.Second) // 设置执行权限 _, err = s.session.Stdin.Write([]byte(fmt.Sprintf("chmod +x %s\n", scriptPath))) if err != nil { return "", err } // 等待权限设置完成 time.Sleep(time.Second) // 执行脚本并将输出发送到WebSocket(在命令中直接删除脚本文件) executeCmd := fmt.Sprintf("bash %s 2>&1; rm -f %s\n", scriptPath, scriptPath) _, err = s.session.Stdin.Write([]byte(executeCmd)) if err != nil { return "", err } // 读取脚本执行输出 output := "" done := make(chan bool) go func() { defer func() { if r := recover(); r != nil { glog.Error(context.Background(), "Script execution goroutine panic:", r) } }() scanner := bufio.NewScanner(s.session.Stdout) for scanner.Scan() { line := scanner.Text() // 检测到脚本执行完成标记则退出 if strings.Contains(line, "#SCRIPT_EXECUTION_COMPLETE#") { break } // 忽略一些可能导致连接断开的输出 if strings.Contains(line, "logout") || strings.Contains(line, "exit") { continue } output += line + "\n" s.sendTerminalOutput(s.session.WebSocket, line) } done <- true }() // 等待脚本执行完成或超时 select { case <-done: return strings.TrimSpace(output), nil case <-time.After(cmdTimeout): return strings.TrimSpace(output), nil } } func (s *ServerHandler) executeFullDeployment() error { s.sendTerminalOutput(s.session.WebSocket, "开始执行完整自动化部署流程...") // 1. 获取并校验下载链接 // filename := config.NewServerService().GetFile() filename, _ := utils.GetLatestLogicFile("public") // fileURL := "http://sun.72wo.cn/" + filename fileURL := "http://" + cool.Config.File.Domain + cool.Config.Address + "/" + filename fileURL = strings.TrimSpace(fileURL) fmt.Println("更新地址", fileURL) if fileURL == "" { return fmt.Errorf("下载链接为空") } // 前置校验:确保链接是合法的URL(包含http/https) if !strings.HasPrefix(fileURL, "http://") && !strings.HasPrefix(fileURL, "https://") { return fmt.Errorf("下载链接格式非法,缺少http/https协议:%s", fileURL) } s.sendTerminalOutput(s.session.WebSocket, fmt.Sprintf("【前置检查】有效下载链接:%s", fileURL)) // 2. 生成目标文件路径 //randomFileName := s.generateRandomFileName() remoteExePath := fmt.Sprintf("%s/%s", defaultWorkDir, filename) remoteWorkDir := defaultWorkDir onlineID := fmt.Sprintf("%d", s.ServerList.OnlineID) fixedScreenSession := "logic" // 3. 定义部署脚本(给每个%s加唯一标记,方便核对) deploymentScriptTpl := ` set -e set -x # ===== 检查并安装screen ===== echo "检查Screen是否已安装..." if command -v screen &> /dev/null; then echo "=== Screen已安装,跳过 ===" else echo "=== Screen未安装,开始安装 ===" if command -v apt &> /dev/null; then apt update -y && apt install -y screen elif command -v yum &> /dev/null; then yum install -y screen elif command -v dnf &> /dev/null; then dnf install -y screen elif command -v pacman &> /dev/null; then pacman -S --noconfirm screen else echo "❌ 不支持的包管理器,无法安装Screen" exit 1 fi command -v screen || { echo "❌ Screen安装失败"; exit 1; } fi #!/bin/bash # 核心:无任何可能阻塞的命令,全程实时输出,100%不卡住 + 彻底清理所有logic会话 set -euo pipefail # ===== 准备下载目录 ===== echo "创建工作目录:%s{work_dir}" mkdir -p "%s{work_dir}" || { echo "❌ 创建目录失败"; exit 1; } # ===== 下载程序 ===== echo "===== 开始下载程序 =====" echo "下载链接:%s{file_url}" echo "目标路径:%s{exe_path}" # 定义要删除文件的目录(根据你的实际路径修改,比如 /usr/local/game/ 或当前目录 .) target_dir="." # 删除所有 logic_ 开头的文件 echo "开始删除 ${target_dir} 目录下 logic_ 开头的文件..." for file in "${target_dir}"/logic_*; do # 检查文件是否存在(避免匹配不到时删除空值) if [ -f "$file" ]; then echo "删除旧文件:$file" rm -f "$file" fi done echo "logic_ 开头的文件删除完成" # ===== 准备下载目录 ===== echo "创建工作目录:%s{work_dir}" mkdir -p "%s{work_dir}" || { echo "❌ 创建目录失败"; exit 1; } # ===== 下载程序 ===== echo "===== 开始下载程序 =====" echo "下载链接:%s{file_url}" echo "目标路径:%s{exe_path}" # 删除旧文件(关键修复:这里要判断目标路径,不是下载链接) echo "删除旧文件:%s{exe_path}" rm -f "%s{exe_path}" echo "开始下载..." DOWNLOAD_SUCCESS=0 if command -v wget >/dev/null 2>&1; then # 正确格式:wget -O 目标路径 下载链接 wget --no-check-certificate --timeout=10 --tries=2 -O "%s{exe_path}" "%s{file_url}" DOWNLOAD_SUCCESS=$? elif command -v curl >/dev/null 2>&1; then # 正确格式:curl -o 目标路径 下载链接 curl -L --connect-timeout 10 --max-time 30 -o "%s{exe_path}" "%s{file_url}" DOWNLOAD_SUCCESS=$? else echo "❌ 无wget/curl,无法下载" exit 1 fi if [ $DOWNLOAD_SUCCESS -ne 0 ]; then echo "❌ 下载失败,退出码:$DOWNLOAD_SUCCESS" exit 1 fi # 验证文件 if [ -f "%s{exe_path}" ] && [ -s "%s{exe_path}" ]; then echo "=== 文件下载完成 ===" ls -la "%s{exe_path}" # 检查文件大小(至少1KB) FILE_SIZE=$(stat -c%s "%s{exe_path}" 2>/dev/null || stat -f%z "%s{exe_path}" 2>/dev/null) [ "$FILE_SIZE" -lt 1024 ] && { echo "❌ 文件太小($FILE_SIZE字节)"; exit 1; } else echo "❌ 文件下载失败或为空" exit 1 fi # ===== 启动新程序 ===== echo "设置执行权限:%s{exe_path}" chmod +x "%s{exe_path}" || { echo "❌ 设置权限失败"; exit 1; } echo "启动Screen会话[%s{screen_name}]..." screen -dmS "%s{screen_name}" bash -c '"%s{exe_path}" -id=%s{online_id} 2>&1 | tee -a "$HOME/run_%s{randomFileName}.log" stty intr ^C' sleep 2 if screen -ls | grep -q "%s{screen_name}"; then echo "✅ 程序启动成功!会话名称:%s{screen_name}" screen -ls else echo "❌ 程序启动失败,未创建[%s{screen_name}]会话" screen -ls exit 1 fi echo "#SCRIPT_EXECUTION_COMPLETE#" ` // 4. 定义参数映射(用占位符替换,彻底避免数错顺序) deploymentScript := strings.ReplaceAll(deploymentScriptTpl, "%s{screen_name}", fixedScreenSession) deploymentScript = strings.ReplaceAll(deploymentScript, "%s{work_dir}", remoteWorkDir) deploymentScript = strings.ReplaceAll(deploymentScript, "%s{file_url}", fileURL) deploymentScript = strings.ReplaceAll(deploymentScript, "%s{exe_path}", remoteExePath) deploymentScript = strings.ReplaceAll(deploymentScript, "%s{online_id}", onlineID) deploymentScript = strings.ReplaceAll(deploymentScript, "%s{randomFileName}", filename) // 5. 执行脚本 _, err := s.executeScript(deploymentScript, "full_deployment_"+grand.S(10)) if err != nil { return fmt.Errorf("执行部署脚本失败:%w", err) } // 6. 保存会话名称 config.NewServerService().SetServerScreen(s.ServerList.OnlineID, filename) return nil } // executeCommand 执行单个命令并返回输出 func (s *ServerHandler) executeCommand(command string) (string, error) { return s.executeScript(command, "cmd_"+grand.S(8)) } // ---------------- 主流程:OnOpen(严格顺序执行) ---------------- func (s *ServerHandler) OnOpen(socket *gws.Conn) { // 1. 建立SSH连接 sshConfig := &ssh.ClientConfig{ User: s.ServerList.Account, Auth: []ssh.AuthMethod{ssh.Password(s.ServerList.Password)}, HostKeyCallback: ssh.InsecureIgnoreHostKey(), Timeout: 30 * time.Second, } // 连接服务器(重试) var sshClient *ssh.Client var err error for retry := 0; retry < 2; retry++ { sshClient, err = ssh.Dial("tcp", s.ServerList.LoginAddr, sshConfig) if err == nil { break } time.Sleep(3 * time.Second) } if err != nil { s.sendError(socket, "SSH连接失败:"+err.Error()) return } // 创建会话(重试) var session *ssh.Session for retry := 0; retry < 2; retry++ { session, err = sshClient.NewSession() if err == nil { break } time.Sleep(3 * time.Second) } if err != nil { session.Close() sshClient.Close() s.sendError(socket, "创建SSH会话失败:"+err.Error()) return } // 配置PTY(通用配置) modes := ssh.TerminalModes{ ssh.ECHO: 1, ssh.TTY_OP_ISPEED: 115200, ssh.TTY_OP_OSPEED: 115200, } err = session.RequestPty("vt100", 100, 80, modes) if err != nil { session.Close() sshClient.Close() s.sendError(socket, "PTY请求失败:"+err.Error()) return } // 获取Stdin/Stdout stdin, err := session.StdinPipe() if err != nil { session.Close() sshClient.Close() s.sendError(socket, "获取Stdin失败:"+err.Error()) return } stdout, err := session.StdoutPipe() if err != nil { session.Close() sshClient.Close() s.sendError(socket, "获取Stdout失败:"+err.Error()) return } // 启动Shell err = session.Shell() if err != nil { session.Close() sshClient.Close() s.sendError(socket, "启动Shell失败:"+err.Error()) return } // 初始化会话 s.session = &TerminalSession{ WebSocket: socket, SSHClient: sshClient, SSHSession: session, Stdin: stdin, Stdout: stdout, Ready: true, outputBuf: make(chan string, 2048), // 增大缓存 closed: atomic.Bool{}, } // 启动输出转发协程 s.startOutputForwarding() s.sendSuccess(socket, "SSH连接成功,会话已就绪") // ---------------- 严格按顺序执行自动化部署 ---------------- if s.isinstall == 1 { // 执行完整的自动化部署脚本 err := s.executeFullDeployment() if err != nil { s.sendError(socket, "自动化部署失败: "+err.Error()) return } s.sendSuccess(socket, "自动化部署完成") } } // startOutputForwarding 启动输出转发协程 func (s *ServerHandler) startOutputForwarding() { if s.session == nil { return } go func() { defer func() { if r := recover(); r != nil { glog.Error(context.Background(), "Output forwarding goroutine panic:", r) } }() scanner := bufio.NewScanner(s.session.Stdout) for scanner.Scan() { line := scanner.Text() s.sendTerminalOutput(s.session.WebSocket, line+"\r\n") } if err := scanner.Err(); err != nil && s.session != nil { glog.Error(context.Background(), "读取 SSH 输出失败:", err) s.sendError(s.session.WebSocket, "读取 SSH 输出失败: "+err.Error()) } }() } // ---------------- 基础函数 ---------------- func (s *ServerHandler) OnMessage(socket *gws.Conn, gwsmessage *gws.Message) { if s.session == nil || s.session.Stdin == nil { s.sendError(socket, "SSH 会话未建立") return } _, err := s.session.Stdin.Write([]byte(gwsmessage.Data.Bytes())) if err != nil { s.sendError(socket, "写入消息失败:"+err.Error()) } } func (s *ServerHandler) OnClose(socket *gws.Conn, err error) { //glog.Debug(context.Background(), "连接断开:", err) if s.session != nil { s.session.closed.Store(true) if s.session.SSHSession != nil { s.session.SSHSession.Close() } if s.session.SSHClient != nil { s.session.SSHClient.Close() } s.session = nil } } // 发送终端输出 func (s *ServerHandler) sendTerminalOutput(ws *gws.Conn, output string) { msg := WebSSHMessage{Type: "output", Payload: output} data, _ := json.Marshal(msg) ws.WriteMessage(gws.OpcodeText, data) } // 发送成功消息 func (s *ServerHandler) sendSuccess(ws *gws.Conn, msg string) { res := WebSSHMessage{Type: "success", Payload: msg} data, _ := json.Marshal(res) ws.WriteMessage(gws.OpcodeText, data) } // 发送错误消息 func (s *ServerHandler) sendError(ws *gws.Conn, msg string) { res := WebSSHMessage{Type: "error", Payload: msg} data, _ := json.Marshal(res) ws.WriteMessage(gws.OpcodeText, data) }