Logan.Li
ee59706ab9
- 类型精简为发布流程所需 4 种：SSH 密钥 / Docker 认证 / Git 仓库 / Kubernetes（旧 password/kubeconfig/aksk 仅作历史数据展示，新建不提供） - 类型化表单替代通用 JSON textarea（aiops 模式）： ssh=私钥/用户/端口/密码；docker=Registry/用户/Token； git=仓库/用户/Token；编辑时留空=不修改，类型不可改 - Kubernetes 凭证双模式（LokiStack 模式），默认 Token（推荐）： apiServer+命名空间+Token，或整份 Kubeconfig+context - 命名空间长期授权引导：折叠块内一键复制 kubectl 命令 （Namespace+SA+命名空间级 Role/RoleBinding+长期 Token Secret）， 命名空间随表单动态注入；附最小权限安全说明 - 后端摘要支持 kubernetes 类型（K8s Token ns@server / Kubeconfig）
762 lines
23 KiB
Go
762 lines
23 KiB
Go
package handler
|
||
|
||
import (
|
||
"crypto/hmac"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"fmt"
|
||
"net"
|
||
"net/http"
|
||
"os/exec"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
|
||
"flowstack/internal/crypto"
|
||
"flowstack/internal/database"
|
||
"flowstack/internal/model"
|
||
)
|
||
|
||
// dialTCP 真实连通性测试：TCP 拨号 5 秒超时
|
||
func dialTCP(addr string) error {
|
||
conn, err := net.DialTimeout("tcp", addr, 5*time.Second)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return conn.Close()
|
||
}
|
||
|
||
// ============================================================
|
||
// 主机 /api/hosts
|
||
// ============================================================
|
||
|
||
func ListHosts(c *gin.Context) {
|
||
var list []model.Host
|
||
q := database.DB.Order("updated_at DESC")
|
||
if tag := c.Query("tag"); tag != "" {
|
||
q = q.Where("tags LIKE ?", "%"+tag+"%")
|
||
}
|
||
q.Find(&list)
|
||
c.JSON(http.StatusOK, list)
|
||
}
|
||
|
||
func CreateHost(c *gin.Context) {
|
||
var h model.Host
|
||
if err := c.ShouldBindJSON(&h); err != nil || h.Name == "" || h.IP == "" {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": "名称和 IP 不能为空"})
|
||
return
|
||
}
|
||
h.ID = 0
|
||
if err := database.DB.Create(&h).Error; err != nil {
|
||
c.JSON(http.StatusConflict, gin.H{"error": "创建失败：名称可能已存在"})
|
||
return
|
||
}
|
||
c.JSON(http.StatusOK, h)
|
||
}
|
||
|
||
func UpdateHost(c *gin.Context) {
|
||
var h model.Host
|
||
if err := database.DB.First(&h, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "host not found"})
|
||
return
|
||
}
|
||
var body map[string]any
|
||
_ = c.ShouldBindJSON(&body)
|
||
for _, k := range []string{"name", "ip", "port", "username", "tags", "os", "remark"} {
|
||
if v, ok := body[k]; ok {
|
||
database.DB.Model(&h).Update(k, v)
|
||
}
|
||
}
|
||
database.DB.First(&h, h.ID)
|
||
c.JSON(http.StatusOK, h)
|
||
}
|
||
|
||
func DeleteHost(c *gin.Context) {
|
||
database.DB.Delete(&model.Host{}, c.Param("id"))
|
||
c.JSON(http.StatusOK, gin.H{"success": true})
|
||
}
|
||
|
||
// TestHost POST /api/hosts/:id/test —— 真实 TCP 拨号 SSH 端口
|
||
func TestHost(c *gin.Context) {
|
||
var h model.Host
|
||
if err := database.DB.First(&h, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "host not found"})
|
||
return
|
||
}
|
||
start := time.Now()
|
||
err := dialTCP(fmt.Sprintf("%s:%d", h.IP, h.SSHPort))
|
||
latency := time.Since(start).Milliseconds()
|
||
status := "online"
|
||
if err != nil {
|
||
status = "offline"
|
||
}
|
||
database.DB.Model(&h).Update("status", status)
|
||
c.JSON(http.StatusOK, gin.H{
|
||
"success": err == nil, "status": status, "latencyMs": latency,
|
||
"message": map[bool]string{true: fmt.Sprintf("SSH 端口可达 (%dms)", latency), false: "连接失败: " + err.Error()}[err == nil],
|
||
})
|
||
}
|
||
|
||
// ============================================================
|
||
// 集群 /api/clusters
|
||
// ============================================================
|
||
|
||
func ListClusters(c *gin.Context) {
|
||
var list []model.K8sCluster
|
||
database.DB.Order("updated_at DESC").Find(&list)
|
||
for i := range list {
|
||
list[i].KubeconfigSet = list[i].Kubeconfig != ""
|
||
list[i].Kubeconfig = ""
|
||
}
|
||
c.JSON(http.StatusOK, list)
|
||
}
|
||
|
||
func CreateCluster(c *gin.Context) {
|
||
var raw map[string]any
|
||
if err := c.ShouldBindJSON(&raw); err != nil {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
name, _ := raw["name"].(string)
|
||
if name == "" {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": "名称不能为空"})
|
||
return
|
||
}
|
||
cl := model.K8sCluster{Name: name}
|
||
cl.Context, _ = raw["context"].(string)
|
||
cl.APIServer, _ = raw["apiServer"].(string)
|
||
cl.Version, _ = raw["version"].(string)
|
||
if v, ok := raw["nodes"].(float64); ok {
|
||
cl.NodeCount = int(v)
|
||
}
|
||
if v, ok := raw["namespaces"].(string); ok {
|
||
cl.Namespaces = v
|
||
} else if ns, ok := raw["namespaces"].([]any); ok {
|
||
if b, err := json.Marshal(ns); err == nil {
|
||
cl.Namespaces = string(b)
|
||
}
|
||
}
|
||
if kubeconfig, _ := raw["kubeconfig"].(string); kubeconfig != "" {
|
||
enc, err := crypto.EncryptField(kubeconfig)
|
||
if err != nil {
|
||
c.JSON(http.StatusInternalServerError, gin.H{"error": "kubeconfig 加密失败"})
|
||
return
|
||
}
|
||
cl.Kubeconfig = enc
|
||
}
|
||
if err := database.DB.Create(&cl).Error; err != nil {
|
||
c.JSON(http.StatusConflict, gin.H{"error": "创建失败：名称可能已存在"})
|
||
return
|
||
}
|
||
cl.KubeconfigSet = cl.Kubeconfig != ""
|
||
cl.Kubeconfig = ""
|
||
c.JSON(http.StatusOK, cl)
|
||
}
|
||
|
||
func UpdateCluster(c *gin.Context) {
|
||
var cl model.K8sCluster
|
||
if err := database.DB.First(&cl, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"})
|
||
return
|
||
}
|
||
var raw map[string]any
|
||
if err := c.ShouldBindJSON(&raw); err != nil {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
updates := map[string]any{}
|
||
for _, k := range []string{"name", "context", "apiServer", "version", "nodes", "namespaces"} {
|
||
if v, ok := raw[k]; ok {
|
||
updates[k] = v
|
||
}
|
||
}
|
||
// kubeconfig 只写不读：非空才更新
|
||
if kubeconfig, ok := raw["kubeconfig"].(string); ok && kubeconfig != "" {
|
||
enc, err := crypto.EncryptField(kubeconfig)
|
||
if err != nil {
|
||
c.JSON(http.StatusInternalServerError, gin.H{"error": "kubeconfig 加密失败"})
|
||
return
|
||
}
|
||
updates["kubeconfig"] = enc
|
||
}
|
||
if len(updates) > 0 {
|
||
database.DB.Model(&cl).Updates(updates)
|
||
}
|
||
database.DB.First(&cl, cl.ID)
|
||
cl.KubeconfigSet = cl.Kubeconfig != ""
|
||
cl.Kubeconfig = ""
|
||
c.JSON(http.StatusOK, cl)
|
||
}
|
||
|
||
func DeleteCluster(c *gin.Context) {
|
||
database.DB.Delete(&model.K8sCluster{}, c.Param("id"))
|
||
c.JSON(http.StatusOK, gin.H{"success": true})
|
||
}
|
||
|
||
// TestCluster POST /api/clusters/:id/test —— 拨号 APIServer（或 kubeconfig 中的 server）
|
||
func TestCluster(c *gin.Context) {
|
||
var cl model.K8sCluster
|
||
if err := database.DB.First(&cl, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"})
|
||
return
|
||
}
|
||
target := cl.APIServer
|
||
if cl.Kubeconfig != "" && crypto.IsEncrypted(cl.Kubeconfig) {
|
||
if plain, err := crypto.DecryptField(cl.Kubeconfig); err == nil {
|
||
for _, line := range strings.Split(plain, "\n") {
|
||
if trimmed := strings.TrimSpace(line); strings.HasPrefix(trimmed, "server:") {
|
||
target = strings.TrimSpace(strings.TrimPrefix(trimmed, "server:"))
|
||
break
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if target == "" {
|
||
c.JSON(http.StatusBadRequest, gin.H{"success": false, "message": "未配置 APIServer 或 kubeconfig"})
|
||
return
|
||
}
|
||
host := strings.TrimPrefix(strings.TrimPrefix(target, "https://"), "http://")
|
||
err := dialTCP(host)
|
||
status := "connected"
|
||
if err != nil {
|
||
status = "disconnected"
|
||
}
|
||
now := time.Now()
|
||
database.DB.Model(&cl).Updates(map[string]any{"status": status, "last_heartbeat": &now})
|
||
c.JSON(http.StatusOK, gin.H{
|
||
"success": err == nil, "status": status,
|
||
"message": map[bool]string{true: "APIServer 可达", false: "连接失败: " + err.Error()}[err == nil],
|
||
})
|
||
}
|
||
|
||
// ============================================================
|
||
// 凭证 /api/credentials（内容只写不读）
|
||
// ============================================================
|
||
|
||
func ListCredentials(c *gin.Context) {
|
||
var list []model.Credential
|
||
q := database.DB.Order("updated_at DESC")
|
||
if t := c.Query("type"); t != "" && t != "all" {
|
||
q = q.Where("type = ?", t)
|
||
}
|
||
q.Find(&list)
|
||
for i := range list {
|
||
list[i].ContentSet = list[i].Content != ""
|
||
list[i].Content = ""
|
||
}
|
||
c.JSON(http.StatusOK, list)
|
||
}
|
||
|
||
func CreateCredential(c *gin.Context) {
|
||
var raw map[string]any
|
||
if err := c.ShouldBindJSON(&raw); err != nil {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
name, _ := raw["name"].(string)
|
||
typ, _ := raw["type"].(string)
|
||
content, _ := raw["content"].(string)
|
||
if name == "" || typ == "" {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": "名称和类型不能为空"})
|
||
return
|
||
}
|
||
cred := model.Credential{Name: name, Type: typ}
|
||
if scope, ok := raw["scope"].(string); ok {
|
||
cred.Scope = scope
|
||
}
|
||
if content != "" {
|
||
enc, err := crypto.EncryptField(content)
|
||
if err != nil {
|
||
c.JSON(http.StatusInternalServerError, gin.H{"error": "加密失败"})
|
||
return
|
||
}
|
||
cred.Content = enc
|
||
cred.Summary = summarizeCredential(typ, content)
|
||
}
|
||
if err := database.DB.Create(&cred).Error; err != nil {
|
||
c.JSON(http.StatusConflict, gin.H{"error": "创建失败：名称可能已存在"})
|
||
return
|
||
}
|
||
cred.ContentSet = cred.Content != ""
|
||
cred.Content = ""
|
||
c.JSON(http.StatusOK, cred)
|
||
}
|
||
|
||
// summarizeCredential 从明文 JSON 中提取脱敏摘要（不落明文）
|
||
func summarizeCredential(typ, contentJSON string) string {
|
||
var m map[string]any
|
||
if err := json.Unmarshal([]byte(contentJSON), &m); err != nil {
|
||
return typ + " 凭证"
|
||
}
|
||
pick := func(keys ...string) string {
|
||
for _, k := range keys {
|
||
if v, ok := m[k].(string); ok && v != "" {
|
||
return k + "=" + crypto.Mask(v)
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
switch typ {
|
||
case "ssh":
|
||
return typ + " " + pick("ssh_user", "username", "private_key")
|
||
case "docker":
|
||
return typ + " " + pick("server") + " " + pick("username")
|
||
case "git":
|
||
return typ + " " + pick("url", "username")
|
||
case "kubernetes":
|
||
mode, _ := m["mode"].(string)
|
||
if mode == "kubeconfig" {
|
||
return "K8s Kubeconfig"
|
||
}
|
||
return "K8s Token " + pick("namespace") + "@" + pick("apiServer")
|
||
case "kubeconfig":
|
||
return typ + " (kubeconfig)"
|
||
case "aksk":
|
||
return typ + " " + pick("access_key")
|
||
default:
|
||
if len(m) > 0 {
|
||
return typ + " (" + fmt.Sprint(len(m)) + " 项)"
|
||
}
|
||
return typ + " 凭证"
|
||
}
|
||
}
|
||
|
||
func UpdateCredential(c *gin.Context) {
|
||
var cred model.Credential
|
||
if err := database.DB.First(&cred, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "credential not found"})
|
||
return
|
||
}
|
||
var raw map[string]any
|
||
if err := c.ShouldBindJSON(&raw); err != nil {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
updates := map[string]any{}
|
||
if v, ok := raw["name"].(string); ok && v != "" {
|
||
updates["name"] = v
|
||
}
|
||
if v, ok := raw["scope"].(string); ok {
|
||
updates["scope"] = v
|
||
}
|
||
if content, ok := raw["content"].(string); ok && content != "" {
|
||
enc, err := crypto.EncryptField(content)
|
||
if err != nil {
|
||
c.JSON(http.StatusInternalServerError, gin.H{"error": "加密失败"})
|
||
return
|
||
}
|
||
updates["content"] = enc
|
||
typ, _ := raw["type"].(string)
|
||
if typ == "" {
|
||
typ = cred.Type
|
||
}
|
||
updates["summary"] = summarizeCredential(typ, content)
|
||
}
|
||
if len(updates) > 0 {
|
||
database.DB.Model(&cred).Updates(updates)
|
||
}
|
||
database.DB.First(&cred, cred.ID)
|
||
cred.ContentSet = cred.Content != ""
|
||
cred.Content = ""
|
||
c.JSON(http.StatusOK, cred)
|
||
}
|
||
|
||
func DeleteCredential(c *gin.Context) {
|
||
database.DB.Delete(&model.Credential{}, c.Param("id"))
|
||
c.JSON(http.StatusOK, gin.H{"success": true})
|
||
}
|
||
|
||
// ============================================================
|
||
// 构建节点 /api/build-nodes
|
||
// ============================================================
|
||
|
||
func ListBuildNodes(c *gin.Context) {
|
||
var list []model.BuildNode
|
||
database.DB.Order("updated_at DESC").Find(&list)
|
||
c.JSON(http.StatusOK, list)
|
||
}
|
||
|
||
func CreateBuildNode(c *gin.Context) {
|
||
var n model.BuildNode
|
||
if err := c.ShouldBindJSON(&n); err != nil || n.Name == "" || n.IP == "" {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": "名称和 IP 不能为空"})
|
||
return
|
||
}
|
||
n.ID = 0
|
||
if err := database.DB.Create(&n).Error; err != nil {
|
||
c.JSON(http.StatusConflict, gin.H{"error": "创建失败：名称可能已存在"})
|
||
return
|
||
}
|
||
c.JSON(http.StatusOK, n)
|
||
}
|
||
|
||
func UpdateBuildNode(c *gin.Context) {
|
||
var n model.BuildNode
|
||
if err := database.DB.First(&n, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "build node not found"})
|
||
return
|
||
}
|
||
var raw map[string]any
|
||
_ = c.ShouldBindJSON(&raw)
|
||
for _, k := range []string{"name", "ip", "labels", "maxConcurrency", "workspacePath", "diskInfo", "credentialId", "enabled"} {
|
||
if v, ok := raw[k]; ok {
|
||
database.DB.Model(&n).Update(k, v)
|
||
}
|
||
}
|
||
database.DB.First(&n, n.ID)
|
||
c.JSON(http.StatusOK, n)
|
||
}
|
||
|
||
func DeleteBuildNode(c *gin.Context) {
|
||
database.DB.Delete(&model.BuildNode{}, c.Param("id"))
|
||
c.JSON(http.StatusOK, gin.H{"success": true})
|
||
}
|
||
|
||
// TestBuildNode POST /api/build-nodes/:id/test —— 真实 TCP 拨号
|
||
func TestBuildNode(c *gin.Context) {
|
||
var n model.BuildNode
|
||
if err := database.DB.First(&n, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "build node not found"})
|
||
return
|
||
}
|
||
start := time.Now()
|
||
err := dialTCP(n.IP + ":22")
|
||
latency := time.Since(start).Milliseconds()
|
||
status := "online"
|
||
if err != nil {
|
||
status = "offline"
|
||
}
|
||
now := time.Now()
|
||
database.DB.Model(&n).Updates(map[string]any{"status": status, "last_heartbeat": &now})
|
||
c.JSON(http.StatusOK, gin.H{
|
||
"success": err == nil, "status": status, "latencyMs": latency,
|
||
"message": map[bool]string{true: fmt.Sprintf("节点可达 (%dms)", latency), false: "连接失败: " + err.Error()}[err == nil],
|
||
})
|
||
}
|
||
|
||
// ============================================================
|
||
// 工具 /api/tools（真实检测 + 保守安装）
|
||
// ============================================================
|
||
|
||
func ListTools(c *gin.Context) {
|
||
var list []model.DeployTool
|
||
database.DB.Order("sort_order").Find(&list)
|
||
c.JSON(http.StatusOK, list)
|
||
}
|
||
|
||
func UpdateTool(c *gin.Context) {
|
||
var t model.DeployTool
|
||
if err := database.DB.First(&t, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "tool not found"})
|
||
return
|
||
}
|
||
var raw map[string]any
|
||
_ = c.ShouldBindJSON(&raw)
|
||
for _, k := range []string{"installMode", "version", "path", "installHint", "enabled"} {
|
||
if v, ok := raw[k]; ok {
|
||
database.DB.Model(&t).Update(k, v)
|
||
}
|
||
}
|
||
database.DB.First(&t, t.ID)
|
||
c.JSON(http.StatusOK, t)
|
||
}
|
||
|
||
// runToolCheck 真实执行 CheckCommand，返回输出中的版本号片段
|
||
func runToolCheck(checkCommand string) (version string, out string, err error) {
|
||
cmd := exec.Command("sh", "-c", checkCommand)
|
||
cmd.Stderr = nil
|
||
bytes, err := cmd.Output()
|
||
out = strings.TrimSpace(string(bytes))
|
||
if err != nil {
|
||
// java -version 等把版本打到 stderr
|
||
if exitErr, ok := err.(*exec.ExitError); ok && len(exitErr.Stderr) > 0 {
|
||
out = strings.TrimSpace(string(exitErr.Stderr))
|
||
for _, line := range strings.Split(out, "\n") {
|
||
if strings.Contains(line, "version") {
|
||
return extractVersion(line), out, nil
|
||
}
|
||
}
|
||
}
|
||
return "", out, err
|
||
}
|
||
return extractVersion(out), out, nil
|
||
}
|
||
|
||
// extractVersion 从命令输出中提取首个 x.y.z 形态的版本号
|
||
func extractVersion(s string) string {
|
||
fields := strings.Fields(s)
|
||
inVersion := false
|
||
for i, f := range fields {
|
||
if strings.Contains(f, "version") {
|
||
if i+1 < len(fields) {
|
||
return strings.Trim(fields[i+1], `"`)
|
||
}
|
||
inVersion = true
|
||
continue
|
||
}
|
||
if inVersion {
|
||
return strings.Trim(f, `"`)
|
||
}
|
||
}
|
||
// 兜底：找 x.y.z
|
||
for _, f := range fields {
|
||
if strings.Count(f, ".") >= 1 && f[0] >= '0' && f[0] <= '9' {
|
||
return strings.Trim(f, `"`)
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// CheckTool POST /api/tools/:id/check —— 真实执行检测命令并回填版本
|
||
func CheckTool(c *gin.Context) {
|
||
var t model.DeployTool
|
||
if err := database.DB.First(&t, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "tool not found"})
|
||
return
|
||
}
|
||
version, out, err := runToolCheck(t.CheckCommand)
|
||
now := time.Now()
|
||
updates := map[string]any{"last_check": &now}
|
||
if err == nil && version != "" {
|
||
updates["status"] = "installed"
|
||
updates["installed_version"] = version
|
||
updates["error_msg"] = ""
|
||
} else {
|
||
updates["status"] = "not-installed"
|
||
updates["error_msg"] = "未检测到：" + firstLine(out, err)
|
||
}
|
||
database.DB.Model(&t).Updates(updates)
|
||
database.DB.First(&t, t.ID)
|
||
c.JSON(http.StatusOK, gin.H{"tool": t, "output": out, "detected": version})
|
||
}
|
||
|
||
// InstallTool POST /api/tools/:id/install —— 保守安装：真实检测，检出即装；未检出标记 failed 提示手动安装
|
||
func InstallTool(c *gin.Context) {
|
||
var t model.DeployTool
|
||
if err := database.DB.First(&t, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "tool not found"})
|
||
return
|
||
}
|
||
database.DB.Model(&t).Update("status", "installing")
|
||
version, out, err := runToolCheck(t.CheckCommand)
|
||
now := time.Now()
|
||
updates := map[string]any{"last_check": &now}
|
||
if err == nil && version != "" {
|
||
updates["status"] = "installed"
|
||
updates["installed_version"] = version
|
||
updates["error_msg"] = ""
|
||
c.JSON(http.StatusOK, gin.H{"success": true, "message": "检测到已安装 " + version, "version": version, "output": out})
|
||
} else {
|
||
updates["status"] = "failed"
|
||
updates["error_msg"] = "本机未检测到 " + t.DisplayName + "，保守安装策略不会自动下载，请手动安装（期望版本 " + t.Version + "）"
|
||
c.JSON(http.StatusOK, gin.H{"success": false, "message": updates["error_msg"], "output": firstLine(out, err)})
|
||
}
|
||
database.DB.Model(&t).Updates(updates)
|
||
}
|
||
|
||
func firstLine(out string, err error) string {
|
||
if err != nil {
|
||
return "命令执行失败: " + err.Error()
|
||
}
|
||
for _, l := range strings.Split(out, "\n") {
|
||
if l != "" {
|
||
return l
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// ============================================================
|
||
// Webhook /api/webhooks（HMAC-SHA256 签名）
|
||
// ============================================================
|
||
|
||
func ListWebhooks(c *gin.Context) {
|
||
var list []model.WebhookEndpoint
|
||
database.DB.Order("updated_at DESC").Find(&list)
|
||
for i := range list {
|
||
list[i].SecretSet = list[i].Secret != ""
|
||
list[i].Secret = ""
|
||
}
|
||
c.JSON(http.StatusOK, list)
|
||
}
|
||
|
||
func CreateWebhook(c *gin.Context) {
|
||
var raw map[string]any
|
||
if err := c.ShouldBindJSON(&raw); err != nil {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
name, _ := raw["name"].(string)
|
||
url, _ := raw["url"].(string)
|
||
if name == "" || url == "" {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": "名称和 URL 不能为空"})
|
||
return
|
||
}
|
||
wh := model.WebhookEndpoint{Name: name, URL: url, Active: true}
|
||
if events, ok := raw["events"]; ok {
|
||
if b, err := json.Marshal(events); err == nil {
|
||
wh.Events = string(b)
|
||
}
|
||
}
|
||
if secret, ok := raw["secret"].(string); ok && secret != "" {
|
||
enc, err := crypto.EncryptField(secret)
|
||
if err != nil {
|
||
c.JSON(http.StatusInternalServerError, gin.H{"error": "加密失败"})
|
||
return
|
||
}
|
||
wh.Secret = enc
|
||
}
|
||
if err := database.DB.Create(&wh).Error; err != nil {
|
||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
wh.SecretSet = wh.Secret != ""
|
||
wh.Secret = ""
|
||
c.JSON(http.StatusOK, wh)
|
||
}
|
||
|
||
func UpdateWebhook(c *gin.Context) {
|
||
var wh model.WebhookEndpoint
|
||
if err := database.DB.First(&wh, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "webhook not found"})
|
||
return
|
||
}
|
||
var raw map[string]any
|
||
_ = c.ShouldBindJSON(&raw)
|
||
updates := map[string]any{}
|
||
if v, ok := raw["name"].(string); ok {
|
||
updates["name"] = v
|
||
}
|
||
if v, ok := raw["url"].(string); ok {
|
||
updates["url"] = v
|
||
}
|
||
if v, ok := raw["active"].(bool); ok {
|
||
updates["active"] = v
|
||
}
|
||
if events, ok := raw["events"]; ok {
|
||
if b, err := json.Marshal(events); err == nil {
|
||
updates["events"] = string(b)
|
||
}
|
||
}
|
||
if secret, ok := raw["secret"].(string); ok && secret != "" {
|
||
if enc, err := crypto.EncryptField(secret); err == nil {
|
||
updates["secret"] = enc
|
||
}
|
||
}
|
||
if len(updates) > 0 {
|
||
database.DB.Model(&wh).Updates(updates)
|
||
}
|
||
database.DB.First(&wh, wh.ID)
|
||
wh.SecretSet = wh.Secret != ""
|
||
wh.Secret = ""
|
||
c.JSON(http.StatusOK, wh)
|
||
}
|
||
|
||
func DeleteWebhook(c *gin.Context) {
|
||
database.DB.Delete(&model.WebhookEndpoint{}, c.Param("id"))
|
||
c.JSON(http.StatusOK, gin.H{"success": true})
|
||
}
|
||
|
||
// TestWebhook POST /api/webhooks/:id/test —— 真实 HTTP POST + HMAC-SHA256 签名
|
||
func TestWebhook(c *gin.Context) {
|
||
var wh model.WebhookEndpoint
|
||
if err := database.DB.First(&wh, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "webhook not found"})
|
||
return
|
||
}
|
||
payload, _ := json.Marshal(map[string]any{
|
||
"event": "webhook.test", "timestamp": time.Now().Format(time.RFC3339),
|
||
"message": "FlowStack Webhook 测试消息",
|
||
})
|
||
req, err := http.NewRequest("POST", wh.URL, strings.NewReader(string(payload)))
|
||
if err != nil {
|
||
c.JSON(http.StatusBadRequest, gin.H{"success": false, "message": err.Error()})
|
||
return
|
||
}
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("X-FlowStack-Event", "webhook.test")
|
||
if wh.Secret != "" {
|
||
if secret, derr := crypto.DecryptField(wh.Secret); derr == nil && secret != "" {
|
||
mac := hmac.New(sha256.New, []byte(secret))
|
||
mac.Write(payload)
|
||
req.Header.Set("X-FlowStack-Signature", "sha256="+hex.EncodeToString(mac.Sum(nil)))
|
||
}
|
||
}
|
||
client := &http.Client{Timeout: 10 * time.Second}
|
||
resp, err := client.Do(req)
|
||
now := time.Now()
|
||
if err != nil {
|
||
database.DB.Model(&wh).Updates(map[string]any{"last_status": "failed", "last_triggered_at": &now})
|
||
c.JSON(http.StatusOK, gin.H{"success": false, "message": "发送失败: " + err.Error()})
|
||
return
|
||
}
|
||
defer resp.Body.Close()
|
||
status := "success"
|
||
if resp.StatusCode >= 300 {
|
||
status = "failed"
|
||
}
|
||
database.DB.Model(&wh).Updates(map[string]any{"last_status": status, "last_triggered_at": &now})
|
||
c.JSON(http.StatusOK, gin.H{"success": resp.StatusCode < 300, "statusCode": resp.StatusCode, "message": fmt.Sprintf("响应 %d %s", resp.StatusCode, http.StatusText(resp.StatusCode))})
|
||
}
|
||
|
||
// NotifyWebhookEvent 流水线事件发生后通知订阅了该事件的 Webhook（异步调用）
|
||
func NotifyWebhookEvent(event string, data map[string]any) {
|
||
go func() {
|
||
var endpoints []model.WebhookEndpoint
|
||
database.DB.Where("active = ?", true).Find(&endpoints)
|
||
payload, _ := json.Marshal(map[string]any{"event": event, "timestamp": time.Now().Format(time.RFC3339), "data": data})
|
||
client := &http.Client{Timeout: 10 * time.Second}
|
||
for _, wh := range endpoints {
|
||
if !strings.Contains(wh.Events, event) {
|
||
continue
|
||
}
|
||
req, err := http.NewRequest("POST", wh.URL, strings.NewReader(string(payload)))
|
||
if err != nil {
|
||
continue
|
||
}
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("X-FlowStack-Event", event)
|
||
if wh.Secret != "" {
|
||
if secret, derr := crypto.DecryptField(wh.Secret); derr == nil && secret != "" {
|
||
mac := hmac.New(sha256.New, []byte(secret))
|
||
mac.Write(payload)
|
||
req.Header.Set("X-FlowStack-Signature", "sha256="+hex.EncodeToString(mac.Sum(nil)))
|
||
}
|
||
}
|
||
status := "success"
|
||
if resp, err := client.Do(req); err != nil || resp.StatusCode >= 300 {
|
||
status = "failed"
|
||
if resp != nil {
|
||
resp.Body.Close()
|
||
}
|
||
}
|
||
now := time.Now()
|
||
database.DB.Model(&wh).Updates(map[string]any{"last_status": status, "last_triggered_at": &now})
|
||
}
|
||
}()
|
||
}
|
||
|
||
// InboundPipelineHook POST /api/hooks/pipeline/:id?token= —— 入站触发（TriggerPanel 的 Webhook URL）
|
||
func InboundPipelineHook(hub *Hub) gin.HandlerFunc {
|
||
return func(c *gin.Context) {
|
||
var p model.Pipeline
|
||
if err := database.DB.First(&p, c.Param("id")).Error; err != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "pipeline not found"})
|
||
return
|
||
}
|
||
token := c.Query("token")
|
||
if p.WebhookToken != "" && token != p.WebhookToken {
|
||
c.JSON(http.StatusUnauthorized, gin.H{"error": "token 校验失败"})
|
||
return
|
||
}
|
||
if p.TriggerMode != "webhook" && p.TriggerMode != "git_push" {
|
||
c.JSON(http.StatusConflict, gin.H{"error": "流水线未启用 Webhook 触发"})
|
||
return
|
||
}
|
||
go triggerPipelineRun(hub, p, "", "", "webhook", "webhook")
|
||
c.JSON(http.StatusOK, gin.H{"success": true, "message": "已触发流水线 " + p.Name})
|
||
}
|
||
}
|