package handlers import ( "encoding/json" "fmt" "net/http" "strconv" "strings" "time" "ezops/internal/models" "ezops/pkg/remoteexec" "github.com/gin-gonic/gin" "gorm.io/gorm" ) // RemoteExecHandler 远程执行处理器 — Go 原生 SSH 批量执行,替代 Ansible // Logan.Li type RemoteExecHandler struct { db *gorm.DB } // NewRemoteExecHandler 创建远程执行处理器 func NewRemoteExecHandler(db *gorm.DB) *RemoteExecHandler { return &RemoteExecHandler{db: db} } // ==================== 远程命令执行 ==================== // BatchExecRequest 批量执行请求 type BatchExecRequest struct { ServerIDs []uint `json:"server_ids" binding:"required"` // 目标服务器 ID 列表 Command string `json:"command" binding:"required"` // 要执行的命令 Timeout int `json:"timeout"` // 超时(秒),默认 300 } // BatchExec 批量远程执行命令 // Logan.Li // POST /api/v1/remote-exec/execute func (h *RemoteExecHandler) BatchExec(c *gin.Context) { var req BatchExecRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) return } // 查询目标服务器 var servers []models.Server if err := h.db.Where("id IN ?", req.ServerIDs).Find(&servers).Error; err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "查询服务器失败"}) return } if len(servers) == 0 { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "未找到目标服务器"}) return } // 构建 SSH 配置 configs := make([]remoteexec.SSHConfig, 0, len(servers)) for _, s := range servers { cfg, err := buildSSHConfig(h.db, s) if err != nil { c.JSON(http.StatusBadRequest, models.Response{ Code: 400, Message: fmt.Sprintf("服务器 %s 认证配置错误: %v", s.Hostname, err), }) return } configs = append(configs, cfg) } timeout := req.Timeout if timeout <= 0 { timeout = 300 } // 并发执行 results := remoteexec.BatchExec(configs, req.Command, time.Duration(timeout)*time.Second, 10) c.JSON(http.StatusOK, models.Response{Code: 200, Data: results}) } // ==================== 文件批量上传 ==================== // BatchUploadRequest 批量上传请求 type BatchUploadRequest struct { ServerIDs []uint `json:"server_ids" binding:"required"` // 目标服务器 ID 列表 LocalPath string `json:"local_path" binding:"required"` // AIOps 主机上的文件路径 RemotePath string `json:"remote_path" binding:"required"` // 远程目标路径 } // BatchUpload 批量上传文件到远程服务器 // Logan.Li // POST /api/v1/remote-exec/upload func (h *RemoteExecHandler) BatchUpload(c *gin.Context) { var req BatchUploadRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) return } if !strings.HasPrefix(req.LocalPath, "/") { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "local_path 必须是绝对路径"}) return } // 查询目标服务器 var servers []models.Server if err := h.db.Where("id IN ?", req.ServerIDs).Find(&servers).Error; err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "查询服务器失败"}) return } if len(servers) == 0 { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "未找到目标服务器"}) return } // 构建 SSH 配置 configs := make([]remoteexec.SSHConfig, 0, len(servers)) for _, s := range servers { cfg, err := buildSSHConfig(h.db, s) if err != nil { c.JSON(http.StatusBadRequest, models.Response{ Code: 400, Message: fmt.Sprintf("服务器 %s 认证配置错误: %v", s.Hostname, err), }) return } configs = append(configs, cfg) } // 并发上传 results := remoteexec.BatchUpload(configs, req.LocalPath, req.RemotePath, 5) c.JSON(http.StatusOK, models.Response{Code: 200, Data: results}) } // ==================== 主机匹配 ==================== // MatchHosts 预览匹配的主机列表(从资产服务器表查询)// Logan.Li // GET /api/v1/remote-exec/hosts?pattern=web*&match_type=prefix func (h *RemoteExecHandler) MatchHosts(c *gin.Context) { pattern := c.Query("pattern") matchType := c.DefaultQuery("match_type", "contains") query := h.db.Model(&models.Server{}) if pattern != "" { switch matchType { case "prefix": query = query.Where("hostname LIKE ? OR ip LIKE ?", pattern+"%", pattern+"%") case "regex": query = query.Where("hostname REGEXP ? OR ip REGEXP ?", pattern, pattern) default: // contains query = query.Where("hostname LIKE ? OR ip LIKE ? OR tags LIKE ?", "%"+pattern+"%", "%"+pattern+"%", "%"+pattern+"%") } } var total int64 query.Count(&total) var items []models.Server query.Order("hostname").Find(&items) // 补充 HasAuth 字段 for i := range items { items[i].HasAuth = items[i].Password != "" || items[i].SSHKey != "" } c.JSON(http.StatusOK, models.Response{Code: 200, Data: models.ListResponse{List: items, Total: total}}) } // ==================== 单机命令执行 ==================== // ExecOnServer 在指定服务器上执行命令并返回结果 // POST /api/v1/remote-exec/server/:id/exec func (h *RemoteExecHandler) ExecOnServer(c *gin.Context) { id, err := strconv.Atoi(c.Param("id")) if err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "无效的服务器 ID"}) return } var req struct { Command string `json:"command" binding:"required"` Timeout int `json:"timeout"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) return } var server models.Server if err := h.db.First(&server, id).Error; err != nil { c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "服务器不存在"}) return } cfg, err := buildSSHConfig(h.db, server) if err != nil { c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "认证配置错误: " + err.Error()}) return } timeout := req.Timeout if timeout <= 0 { timeout = 300 } result, err := remoteexec.Exec(cfg, req.Command, time.Duration(timeout)*time.Second) if err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error(), Data: result}) return } c.JSON(http.StatusOK, models.Response{Code: 200, Data: result}) } // ==================== 辅助函数 ==================== // buildSSHConfig 从 Server 模型构建 SSHConfig。 // 优先使用 server 自带的 Password/SSHKey 字段, // 如果 server 关联了 credential_id,则从凭证表解密获取。 func buildSSHConfig(db *gorm.DB, server models.Server) (remoteexec.SSHConfig, error) { cfg := remoteexec.SSHConfig{ Host: server.IP, Port: server.Port, User: server.Username, } if cfg.Port == 0 { cfg.Port = 22 } if cfg.User == "" { cfg.User = "root" } // 优先使用 server 自带的认证信息 if server.AuthType == "key" && server.SSHKey != "" { cfg.AuthType = remoteexec.AuthKey cfg.KeyData = server.SSHKey return cfg, nil } if server.Password != "" { cfg.AuthType = remoteexec.AuthPassword cfg.Password = server.Password return cfg, nil } // 从凭证表查找关联的 SSH 凭据 // Server 模型目前没有 CredentialID 字段,通过 tags 或 name 匹配 // 未来可扩展 server.credential_id FK return cfg, fmt.Errorf("服务器 %s 未配置 SSH 认证(密码或密钥)", server.Hostname) } // ==================== 定时任务远程执行(对接 CronJob target_type) ==================== // ExecuteCronOnTargets 根据 CronJob 的 target_type 在目标主机上执行脚本。 // 由 automation_cron.go 调用的公共函数。 func ExecuteCronOnTargets(db *gorm.DB, job models.CronJob, scriptType, content string, timeoutSec int) ([]remoteexec.ExecResult, error) { // 解析目标服务器 var targetIDs []uint if job.Targets != "" { json.Unmarshal([]byte(job.Targets), &targetIDs) } if len(targetIDs) == 0 { return nil, fmt.Errorf("cron job %s has no targets", job.Name) } // 查询服务器 var servers []models.Server switch job.TargetType { case "server": db.Where("id IN ?", targetIDs).Find(&servers) case "group": db.Where("id IN ?", targetIDs).Find(&servers) case "tag": // 标签匹配 — 通过 tags JSON 字段模糊查询 tagStr := fmt.Sprintf("%v", targetIDs) db.Where("tags LIKE ?", "%"+tagStr+"%").Find(&servers) default: db.Where("id IN ?", targetIDs).Find(&servers) } if len(servers) == 0 { return nil, fmt.Errorf("no target servers found for cron job %s", job.Name) } // 构建 SSH 配置 configs := make([]remoteexec.SSHConfig, 0, len(servers)) for _, s := range servers { cfg, err := buildSSHConfig(db, s) if err != nil { continue // 跳过认证失败的主机 } configs = append(configs, cfg) } if len(configs) == 0 { return nil, fmt.Errorf("all target servers have auth errors for cron job %s", job.Name) } // 并发执行 if timeoutSec <= 0 { timeoutSec = 300 } results := remoteexec.BatchExec(configs, content, time.Duration(timeoutSec)*time.Second, 5) return results, nil }