diff --git a/sundynix-desktop/frontend/src/lib/api.ts b/sundynix-desktop/frontend/src/lib/api.ts index 9d6e53e..a71f173 100644 --- a/sundynix-desktop/frontend/src/lib/api.ts +++ b/sundynix-desktop/frontend/src/lib/api.ts @@ -375,15 +375,23 @@ export async function approveTask(taskId: string, approved: boolean, opts?: { no if (!res.ok) throw new Error(`approve failed: ${res.status} ${await res.text()}`); } +// tokenQuery 给 EventSource / 下载链接这类**带不了 Authorization 头**的 URL 附上 JWT, +// 后端 AuthFromHeaderOrQuery 从 ?token= 取它做鉴权 + 归属校验(公开 by-id 端点不再裸奔)。 +function tokenQuery(url: string): string { + const t = getToken(); + if (!t) return url; + return url + (url.includes("?") ? "&" : "?") + "token=" + encodeURIComponent(t); +} + // streamTokens: 订阅 SSE /api/v1/tasks/:id/stream,逐 token 回调,done 收尾。 -// 返回关闭函数。注意 EventSource 无法带请求头,但流按 task_id 寻址,无需身份头。 +// 返回关闭函数。EventSource 无法带请求头,故 JWT 走 ?token=(tokenQuery)。 export function streamTokens( taskId: string, onToken: (t: string) => void, onDone: () => void, onError?: (e: unknown) => void, ): () => void { - const es = new EventSource(`${GATEWAY}/api/v1/tasks/${taskId}/stream`); + const es = new EventSource(tokenQuery(`${GATEWAY}/api/v1/tasks/${taskId}/stream`)); es.addEventListener("token", (e) => onToken((e as MessageEvent).data)); es.addEventListener("done", () => { es.close(); @@ -415,7 +423,7 @@ export function streamExec( onDone: () => void, onError?: (e: unknown) => void, ): () => void { - const es = new EventSource(`${GATEWAY}/api/v1/tasks/${taskId}/exec`); + const es = new EventSource(tokenQuery(`${GATEWAY}/api/v1/tasks/${taskId}/exec`)); es.addEventListener("exec", (e) => onEvent(JSON.parse((e as MessageEvent).data) as ExecEvent)); es.addEventListener("done", () => { es.close(); @@ -594,7 +602,7 @@ export function streamIngest( onDone: () => void, onError?: () => void, ): () => void { - const es = new EventSource(`${GATEWAY}/api/v1/kb/ingest/${jobId}/stream`); + const es = new EventSource(tokenQuery(`${GATEWAY}/api/v1/kb/ingest/${jobId}/stream`)); es.addEventListener("progress", (e) => onEvent(JSON.parse((e as MessageEvent).data) as IngestEvent)); es.addEventListener("done", () => { es.close(); @@ -655,7 +663,7 @@ export async function generateReport(id: Identity, topic: string, kb?: string): // reportDownloadUrl: 渲染好的 Word(.docx) 下载地址(兼容旧入口)。 export function reportDownloadUrl(taskId: string): string { - return `${GATEWAY}/api/v1/reports/${taskId}/download`; + return tokenQuery(`${GATEWAY}/api/v1/reports/${taskId}/download`); } // reportExportUrl: 按需导出报告地址(format=docx|md;后端现渲染)。PDF 由前端打印预览生成。 @@ -666,7 +674,7 @@ export function reportFilename(topic: string, id: string, ext: string): string { } export function reportExportUrl(taskId: string, format: "docx" | "md"): string { - return `${GATEWAY}/api/v1/reports/${taskId}/export?format=${format}`; + return tokenQuery(`${GATEWAY}/api/v1/reports/${taskId}/export?format=${format}`); } // setMemory: PUT /api/v1/memory,登记一条用户偏好(→ mcp-go memory_upsert)。 diff --git a/sundynix-gateway/cmd/server/main.go b/sundynix-gateway/cmd/server/main.go index 198be4a..8071130 100644 --- a/sundynix-gateway/cmd/server/main.go +++ b/sundynix-gateway/cmd/server/main.go @@ -146,7 +146,15 @@ func main() { r := router.New(db, cache, bus, blobStore) addr := envOr("GATEWAY_ADDR", ":8080") - srv := &http.Server{Addr: addr, Handler: r} + // 慢读/Slowloris 防护:限制读头/读体时间与头大小。**不设 WriteTimeout**——会掐断 + // SSE 长连(/tasks/:id/stream、/exec)。ReadTimeout 取 60s 容纳文件上传体。 + srv := &http.Server{ + Addr: addr, + Handler: r, + ReadHeaderTimeout: 10 * time.Second, + ReadTimeout: 60 * time.Second, + MaxHeaderBytes: 1 << 20, // 1MB + } // 后台监听;ListenAndServe 在 Shutdown 后返回 ErrServerClosed(正常退出)。 go func() { diff --git a/sundynix-gateway/internal/handler/admin.go b/sundynix-gateway/internal/handler/admin.go index 52f3b8a..ada2c39 100644 --- a/sundynix-gateway/internal/handler/admin.go +++ b/sundynix-gateway/internal/handler/admin.go @@ -18,19 +18,32 @@ import ( "github.com/sundynix/sundynix-shared/secrets" ) +// clampLimit 解析 ?limit= 并夹到 [1, max],非法/≤0 用 def。 +// 必须夹:负数(如 limit=-1)会让 gorm `Limit(-1)` **取消 LIMIT 子句**,对审计/护栏这类 +// 最易膨胀的表变成全表扫 + 深翻分页。 +func clampLimit(v string, def, max int) int { + n, err := strconv.Atoi(v) + if err != nil || n <= 0 { + return def + } + if n > max { + return max + } + return n +} + +// clampOffset 解析 ?offset=,非法/负数归 0。 +func clampOffset(v string) int { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + return 0 +} + // AuditList: GET /api/v1/admin/audit?limit=&offset= —— 敏感操作审计流(倒序,供运维溯源)。 func (h *Handler) AuditList(c *gin.Context) { - limit, offset := 50, 0 - if v := c.Query("limit"); v != "" { - if n, err := strconv.Atoi(v); err == nil { - limit = n - } - } - if v := c.Query("offset"); v != "" { - if n, err := strconv.Atoi(v); err == nil { - offset = n - } - } + limit := clampLimit(c.Query("limit"), 50, 200) + offset := clampOffset(c.Query("offset")) // 筛选下沉到 SQL:此前是前端在当前页 50 条里过滤,翻页外的记录搜不到, // 对审计来说等于给出错误结论。 f := store.AuditFilter{ @@ -55,17 +68,8 @@ func (h *Handler) AuditList(c *gin.Context) { // GuardrailEvents: GET /api/v1/admin/guardrail-events?limit=&offset= —— 护栏命中安全事件流(倒序)。 func (h *Handler) GuardrailEvents(c *gin.Context) { - limit, offset := 50, 0 - if v := c.Query("limit"); v != "" { - if n, err := strconv.Atoi(v); err == nil { - limit = n - } - } - if v := c.Query("offset"); v != "" { - if n, err := strconv.Atoi(v); err == nil { - offset = n - } - } + limit := clampLimit(c.Query("limit"), 50, 200) + offset := clampOffset(c.Query("offset")) rows, err := h.db.ListGuardrailEvents(c.Request.Context(), limit, offset) if err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) diff --git a/sundynix-gateway/internal/handler/billing_pay.go b/sundynix-gateway/internal/handler/billing_pay.go index b386e6e..0ca7e51 100644 --- a/sundynix-gateway/internal/handler/billing_pay.go +++ b/sundynix-gateway/internal/handler/billing_pay.go @@ -6,7 +6,6 @@ import ( "fmt" "log" "net/http" - "strconv" "strings" "time" @@ -373,12 +372,7 @@ func (h *Handler) AdminReconcile(c *gin.Context) { // 跨租户看所有任务(状态/租户/提交人/评测),含 HITL 待审批(status=waiting)。返回列表 + 状态计数。 func (h *Handler) AdminTasks(c *gin.Context) { ctx := c.Request.Context() - limit := 50 - if v := c.Query("limit"); v != "" { - if n, err := strconv.Atoi(v); err == nil && n > 0 { - limit = n - } - } + limit := clampLimit(c.Query("limit"), 50, 200) rows := h.db.AllTasks(ctx, c.Query("status"), c.Query("tenant"), limit) c.JSON(http.StatusOK, gin.H{"tasks": rows, "counts": h.db.TaskStatusCounts(ctx)}) } @@ -404,12 +398,7 @@ func (h *Handler) AdminTaskDetail(c *gin.Context) { // AdminSpaces: GET /api/v1/admin/spaces?limit= —— 全平台空间观测(跨租户)。 func (h *Handler) AdminSpaces(c *gin.Context) { - limit := 200 - if v := c.Query("limit"); v != "" { - if n, err := strconv.Atoi(v); err == nil && n > 0 { - limit = n - } - } + limit := clampLimit(c.Query("limit"), 200, 500) c.JSON(http.StatusOK, gin.H{"spaces": h.db.AllSpaces(c.Request.Context(), limit)}) } diff --git a/sundynix-gateway/internal/handler/completeness_a_test.go b/sundynix-gateway/internal/handler/completeness_a_test.go new file mode 100644 index 0000000..a27b364 --- /dev/null +++ b/sundynix-gateway/internal/handler/completeness_a_test.go @@ -0,0 +1,44 @@ +package handler + +import "testing" + +// A4:limit 必须夹紧——负数(limit=-1 会让 gorm 取消 LIMIT 全表扫)、0、超大都要归位。 +func TestClampLimit(t *testing.T) { + cases := []struct { + v string + def, max int + want int + }{ + {"", 50, 200, 50}, // 空 → 默认 + {"-1", 50, 200, 50}, // 负数 → 默认(关键:堵住全表扫) + {"0", 50, 200, 50}, // 0 → 默认 + {"abc", 50, 200, 50}, // 非法 → 默认 + {"100", 50, 200, 100}, // 合法 → 原值 + {"999", 50, 200, 200}, // 超上界 → 夹到 max + } + for _, tc := range cases { + if got := clampLimit(tc.v, tc.def, tc.max); got != tc.want { + t.Fatalf("clampLimit(%q,%d,%d)=%d want %d", tc.v, tc.def, tc.max, got, tc.want) + } + } + if clampOffset("-5") != 0 || clampOffset("abc") != 0 || clampOffset("7") != 7 { + t.Fatal("clampOffset 应把负数/非法归 0、合法透传") + } +} + +// A1:safeCall 必须兜住 panic,不外抛(否则后台 goroutine 一 panic 崩整个进程)。 +func TestSafeCallRecovers(t *testing.T) { + done := false + func() { + defer func() { + if r := recover(); r != nil { + t.Fatalf("safeCall 未兜住 panic:%v", r) + } + }() + safeCall("test", func() { panic("boom") }) + done = true + }() + if !done { + t.Fatal("safeCall 之后应正常继续") + } +} diff --git a/sundynix-gateway/internal/handler/kb.go b/sundynix-gateway/internal/handler/kb.go index 2755ae5..47464f4 100644 --- a/sundynix-gateway/internal/handler/kb.go +++ b/sundynix-gateway/internal/handler/kb.go @@ -11,8 +11,10 @@ import ( "io" "log" "net/http" + "os" "path/filepath" "regexp" + "strconv" "strings" "time" @@ -230,6 +232,17 @@ func noteName(text string) string { // KbIngestFile: POST /api/v1/kb/ingest_file(multipart)—— 文件入库(异步,返回 job_id)。 // 流水线(解析→切块→向量化→写入)的进度经 sundynix.streams. 回流,UI 用 SSE 看。 +// kbMaxUploadBytes 返回文件入库大小上限(字节)。默认 50MB,可经 KB_MAX_UPLOAD_BYTES 覆盖。 +func kbMaxUploadBytes() int64 { + const def = 50 << 20 // 50MB + if v := os.Getenv("KB_MAX_UPLOAD_BYTES"); v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil && n > 0 { + return n + } + } + return def +} + func (h *Handler) KbIngestFile(c *gin.Context) { kb := c.PostForm("kb") fh, err := c.FormFile("file") @@ -237,17 +250,28 @@ func (h *Handler) KbIngestFile(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": "file required"}) return } + // 大小闸:先看 multipart 头声明的 Size(快速拒绝),再用 LimitReader 兜底防伪造 Size。 + // 否则整文件 io.ReadAll 进内存 = OOM 面。上限经 KB_MAX_UPLOAD_BYTES 配(默认 50MB)。 + max := kbMaxUploadBytes() + if fh.Size > max { + c.JSON(http.StatusRequestEntityTooLarge, gin.H{"error": "文件过大(上限 " + strconv.FormatInt(max/(1<<20), 10) + "MB)"}) + return + } f, err := fh.Open() if err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) return } defer f.Close() - data, err := io.ReadAll(f) + data, err := io.ReadAll(io.LimitReader(f, max+1)) if err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) return } + if int64(len(data)) > max { + c.JSON(http.StatusRequestEntityTooLarge, gin.H{"error": "文件过大(上限 " + strconv.FormatInt(max/(1<<20), 10) + "MB)"}) + return + } _ = h.db.EnsureKB(c.Request.Context(), spaceID(c), userID(c), rawKB(kb), "general") job, err := h.enqueueIngest(c.Request.Context(), spaceID(c), userID(c), rawKB(kb), scopedKB(c, kb), "", fh.Filename, data, "") if err != nil { diff --git a/sundynix-gateway/internal/handler/payment_reconcile.go b/sundynix-gateway/internal/handler/payment_reconcile.go index 756714e..0e09b77 100644 --- a/sundynix-gateway/internal/handler/payment_reconcile.go +++ b/sundynix-gateway/internal/handler/payment_reconcile.go @@ -56,7 +56,7 @@ func pruneQueryMarks() { // StartReconcile 启动掉单补偿定时器(微信渠道未配置时空转,几乎零成本)。随进程生命周期运行, // ctx 取消即退出。返回给调用方保存以便优雅停机时取消。 func (h *Handler) StartReconcile(ctx context.Context) { - go func() { + safeGo("payment-reconcile-ticker", func() { t := time.NewTicker(reconcileInterval) defer t.Stop() for { @@ -64,11 +64,14 @@ func (h *Handler) StartReconcile(ctx context.Context) { case <-ctx.Done(): return case <-t.C: - h.reconcilePending(ctx) - pruneQueryMarks() + // 单轮兜底:某轮 DB/查单 panic 不该终止整个补偿定时器,下一轮继续。 + safeCall("payment-reconcile-tick", func() { + h.reconcilePending(ctx) + pruneQueryMarks() + }) } } - }() + }) log.Printf("[payment] 掉单补偿定时器已启动(每 %s 扫一次 pending 微信单)", reconcileInterval) } diff --git a/sundynix-gateway/internal/handler/report.go b/sundynix-gateway/internal/handler/report.go index 9a9d7ef..b852f8c 100644 --- a/sundynix-gateway/internal/handler/report.go +++ b/sundynix-gateway/internal/handler/report.go @@ -57,6 +57,9 @@ func (h *Handler) GenerateReport(c *gin.Context) { // 生成阶段只存源;此处经 mcp-go report_export 现渲染("导出时再处理")。PDF 由前端打印预览生成。 func (h *Handler) ExportReport(c *gin.Context) { id := c.Param("id") + if !h.requireTaskOwner(c, id) { // 报告按 task_id 寻址:仅提交者可导出 + return + } format := c.DefaultQuery("format", "docx") res, err := h.bus.CallTool(c.Request.Context(), contract.ToolSubjectGo("report_export"), &contract.ToolCall{Tool: "report_export", Args: map[string]any{"task_id": id, "format": format}}) diff --git a/sundynix-gateway/internal/handler/safego.go b/sundynix-gateway/internal/handler/safego.go new file mode 100644 index 0000000..d67dd82 --- /dev/null +++ b/sundynix-gateway/internal/handler/safego.go @@ -0,0 +1,40 @@ +package handler + +import ( + "log" + "net/http" + "runtime/debug" + + "github.com/gin-gonic/gin" +) + +// requireTaskOwner 校验请求者(AuthFromHeaderOrQuery 已注入的 uid)是该 task 的提交者。 +// 用于公开 by-id 端点(SSE 流 / 报告导出):这些资源始终由本人的客户端访问(用户看/导出 +// 自己提交的运行),故按 owner 判权即安全。非本人 → 403,返回 false。 +func (h *Handler) requireTaskOwner(c *gin.Context, taskID string) bool { + uid := userID(c) + if uid == "" || h.db.TaskOwner(c.Request.Context(), taskID) != uid { + c.JSON(http.StatusForbidden, gin.H{"error": "无权访问该资源"}) + return false + } + return true +} + +// safeGo 起一个带 panic 兜底的后台 goroutine:panic 只记日志(含 name + stack)不外抛。 +// 为什么必须有:Go 里未 recover 的 panic 会崩掉**整个进程**,而 gin.Recovery() 只保护 +// 请求 goroutine、不覆盖 handler 派生的后台 goroutine(定时器/推送/探针)。一个后台任务的 +// 意外 panic 不该拖垮整个 gateway、连带所有在途 HTTP。 +func safeGo(name string, fn func()) { + go safeCall(name, fn) +} + +// safeCall 同步执行 fn 并兜底 panic。用于定时器**单轮**内部:单轮 panic 不该终止整个 ticker, +// 兜住后下一轮照常继续(若把 recover 只放在 safeGo 外层,单轮 panic 会让整个循环 goroutine 结束)。 +func safeCall(name string, fn func()) { + defer func() { + if r := recover(); r != nil { + log.Printf("[panic] 后台任务 %q panic 已兜底: %v\n%s", name, r, debug.Stack()) + } + }() + fn() +} diff --git a/sundynix-gateway/internal/handler/status_handler.go b/sundynix-gateway/internal/handler/status_handler.go index 906287b..1fe624a 100644 --- a/sundynix-gateway/internal/handler/status_handler.go +++ b/sundynix-gateway/internal/handler/status_handler.go @@ -75,58 +75,68 @@ func (h *Handler) AdminStatus(c *gin.Context) { // 1) mcp-go health → milvus / neo4j 基建灯 go func() { defer wg.Done() - ctx, cancel := context.WithTimeout(parent, probeTimeout) - defer cancel() - if res, err := h.bus.CallTool(ctx, contract.ToolSubjectGo("health"), - &contract.ToolCall{Tool: "health"}); err == nil && res != nil && res.OK { - var sub map[string]bool - if json.Unmarshal([]byte(res.Content), &sub) == nil { - milvus, neo4j = sub["milvus"], sub["neo4j"] - ftDisk = sub["fulltext_disk"] + safeCall("status-probe-mcpgo-health", func() { + ctx, cancel := context.WithTimeout(parent, probeTimeout) + defer cancel() + if res, err := h.bus.CallTool(ctx, contract.ToolSubjectGo("health"), + &contract.ToolCall{Tool: "health"}); err == nil && res != nil && res.OK { + var sub map[string]bool + if json.Unmarshal([]byte(res.Content), &sub) == nil { + milvus, neo4j = sub["milvus"], sub["neo4j"] + ftDisk = sub["fulltext_disk"] + } } - } + }) }() // 2) mcp-go list_tools → 在线判定 + 工具清单 go func() { defer wg.Done() - goUp, goTools, goLatency = h.probeTools(parent, contract.ToolSubjectGo("list_tools")) + safeCall("status-probe-mcpgo-tools", func() { + goUp, goTools, goLatency = h.probeTools(parent, contract.ToolSubjectGo("list_tools")) + }) }() // 3) mcp-py list_tools → 在线判定 + 工具清单 go func() { defer wg.Done() - pyUp, pyTools, pyLatency = h.probeTools(parent, contract.ToolSubjectPy("list_tools")) + safeCall("status-probe-mcppy-tools", func() { + pyUp, pyTools, pyLatency = h.probeTools(parent, contract.ToolSubjectPy("list_tools")) + }) }() // 4) dispatcher 心跳 → 在线判定 + 模型/运行时长 go func() { defer wg.Done() - ctx, cancel := context.WithTimeout(parent, probeTimeout) - defer cancel() - start := time.Now() - if data, err := h.bus.Ping(ctx, contract.SubjectHealthDispatcher); err == nil { - dispUp = true - dispLatency = int(time.Since(start).Milliseconds()) - var st struct { - Model string `json:"model"` - Ready bool `json:"ready"` - UptimeS int `json:"uptime_s"` + safeCall("status-probe-dispatcher", func() { + ctx, cancel := context.WithTimeout(parent, probeTimeout) + defer cancel() + start := time.Now() + if data, err := h.bus.Ping(ctx, contract.SubjectHealthDispatcher); err == nil { + dispUp = true + dispLatency = int(time.Since(start).Milliseconds()) + var st struct { + Model string `json:"model"` + Ready bool `json:"ready"` + UptimeS int `json:"uptime_s"` + } + if json.Unmarshal(data, &st) == nil { + dispDetail = dispatcherDetail(st.Model, st.Ready, st.UptimeS) + } } - if json.Unmarshal(data, &st) == nil { - dispDetail = dispatcherDetail(st.Model, st.Ready, st.UptimeS) - } - } + }) }() // 5) 基建活性探针:PG / Redis / MinIO 实时 ping(非仅启动标志,能反映中途掉线)。 go func() { defer wg.Done() - ctx, cancel := context.WithTimeout(parent, probeTimeout) - defer cancel() - pgUp = h.db.Ping(ctx) - redisUp = h.cache.Ping(ctx) - minioUp = h.blob != nil && h.blob.Ping(ctx) + safeCall("status-probe-infra", func() { + ctx, cancel := context.WithTimeout(parent, probeTimeout) + defer cancel() + pgUp = h.db.Ping(ctx) + redisUp = h.cache.Ping(ctx) + minioUp = h.blob != nil && h.blob.Ping(ctx) + }) }() wg.Wait() diff --git a/sundynix-gateway/internal/handler/subscription_tick.go b/sundynix-gateway/internal/handler/subscription_tick.go index cbe1edf..95df06e 100644 --- a/sundynix-gateway/internal/handler/subscription_tick.go +++ b/sundynix-gateway/internal/handler/subscription_tick.go @@ -16,19 +16,20 @@ const subTickInterval = 10 * time.Minute // StartSubscriptionTicker 随进程生命周期运行;多实例并发也安全(发放靠 ledger 唯一索引幂等)。 func (h *Handler) StartSubscriptionTicker(ctx context.Context) { - go func() { + safeGo("subscription-ticker", func() { t := time.NewTicker(subTickInterval) defer t.Stop() - h.tickSubscriptions(ctx) // 启动即跑一次,把停机期间欠的补上 + // 单轮兜底:某轮 panic 不该终止整个定时器,下一轮继续(漏发的下轮补发逻辑兜住)。 + safeCall("subscription-tick", func() { h.tickSubscriptions(ctx) }) // 启动即跑一次,补停机期间欠的 for { select { case <-ctx.Done(): return case <-t.C: - h.tickSubscriptions(ctx) + safeCall("subscription-tick", func() { h.tickSubscriptions(ctx) }) } } - }() + }) log.Printf("[sub] 订阅推进定时器已启动(每 %s 扫一次)", subTickInterval) } diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go index fb6b9ba..9a638f3 100644 --- a/sundynix-gateway/internal/handler/task_handler.go +++ b/sundynix-gateway/internal/handler/task_handler.go @@ -261,6 +261,9 @@ func (h *Handler) ApproveTask(c *gin.Context) { // 优先从 Redis Stream 读(可回放 + 断点续传,根治连晚/重连丢 token);Redis 降级时回退 live NATS。 func (h *Handler) StreamTask(c *gin.Context) { taskID := c.Param("id") + if !h.requireTaskOwner(c, taskID) { // 归属校验须在写 SSE 头之前 + return + } c.Writer.Header().Set("Content-Type", "text/event-stream") c.Writer.Header().Set("Cache-Control", "no-cache") c.Writer.Header().Set("Connection", "keep-alive") @@ -370,6 +373,9 @@ func (h *Handler) Health(c *gin.Context) { // 优先从 Redis Stream 读(可回放 + 断点续传,根治连晚/刷新重连丢轨迹事件);Redis 降级时回退 live NATS。 func (h *Handler) StreamExec(c *gin.Context) { taskID := c.Param("id") + if !h.requireTaskOwner(c, taskID) { + return + } c.Writer.Header().Set("Content-Type", "text/event-stream") c.Writer.Header().Set("Cache-Control", "no-cache") c.Writer.Header().Set("Connection", "keep-alive") diff --git a/sundynix-gateway/internal/handler/wechat_notify.go b/sundynix-gateway/internal/handler/wechat_notify.go index 7eae64d..544d222 100644 --- a/sundynix-gateway/internal/handler/wechat_notify.go +++ b/sundynix-gateway/internal/handler/wechat_notify.go @@ -22,7 +22,7 @@ const wxNotifyTimeout = 10 * time.Second // 绝不影响入账主流程(钱已收、账已记,通知发不发都不能回滚)。 // 仅在 MarkOrderPaid 返回 changed=true(首次到账)时调用,避免重复回调重复推送。 func (h *Handler) notifyOrderPaid(orderID string) { - go func() { + safeGo("wx-notify-order-paid", func() { ctx, cancel := context.WithTimeout(context.Background(), wxNotifyTimeout) defer cancel() @@ -46,7 +46,7 @@ func (h *Handler) notifyOrderPaid(orderID string) { if err := wechat.SendCustomText(ctx, token, u.WechatOpenID, h.composePaidMessage(ctx, o)); err != nil { log.Printf("[wxnotify] 支付回执推送失败 order=%s: %v", orderID, err) } - }() + }) } // composePaidMessage 按订单类型拼回执文案,并带上当前积分余额。 diff --git a/sundynix-gateway/internal/middleware/auth_query.go b/sundynix-gateway/internal/middleware/auth_query.go new file mode 100644 index 0000000..4636106 --- /dev/null +++ b/sundynix-gateway/internal/middleware/auth_query.go @@ -0,0 +1,37 @@ +package middleware + +import ( + "net/http" + "strings" + + "github.com/gin-gonic/gin" + + "github.com/sundynix/sundynix-gateway/internal/auth" +) + +// AuthFromHeaderOrQuery 是给「公开 by-id」端点(SSE 流 / 报告下载)用的强制鉴权。 +// 这些端点由 EventSource / 发起,**带不了 Authorization 头**——此前只靠随机 +// task_id 寻址、无 authz、无租户过滤(capability-URL,泄露即无第二道门)。 +// 本中间件先读 Bearer 头、没有再读 `?token=` / `?access_token=` 查询参数(EventSource 能带 query), +// 校验通过注入 CtxUserID(同 Auth),失败即 401。归属校验由各 handler 用 uid 再做(见 requireTaskOwner)。 +func AuthFromHeaderOrQuery() gin.HandlerFunc { + return func(c *gin.Context) { + tok := "" + if h := c.GetHeader("Authorization"); strings.HasPrefix(h, "Bearer ") { + tok = strings.TrimSpace(h[len("Bearer "):]) + } + if tok == "" { + tok = c.Query("token") + } + if tok == "" { + tok = c.Query("access_token") + } + uid, err := auth.Parse(tok) + if err != nil || uid == "" { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "需要登录"}) + return + } + c.Set(CtxUserID, uid) + c.Next() + } +} diff --git a/sundynix-gateway/internal/middleware/guardrail.go b/sundynix-gateway/internal/middleware/guardrail.go index 70b5de7..e2e73f7 100644 --- a/sundynix-gateway/internal/middleware/guardrail.go +++ b/sundynix-gateway/internal/middleware/guardrail.go @@ -71,9 +71,9 @@ func recordGuardrail(c *gin.Context, db *store.Postgres, kind, reason string, si // RateLimit 基于 Redis 的会话级限流(每分钟上限)。 // 限流键:**已认证用户优先按 uid,未认证按客户端 IP** —— 企业网多人共享出口 IP 不再互相拖累, // 单用户换 IP 也绕不过。须挂在 Auth 之后(否则取不到 uid)。上限经 RATE_LIMIT_PER_MIN 配置 -// (缺省 120);压测可调高。Redis 降级时始终放行,不阻断业务。 +// (缺省 120);压测可调高。**Redis 降级时回落进程内兜底限流(fail-safe),不再完全放行。** func RateLimit(cache *store.Redis) gin.HandlerFunc { - perMinute := int64(envInt("RATE_LIMIT_PER_MIN", 120)) + perMinute := envInt("RATE_LIMIT_PER_MIN", 120) return func(c *gin.Context) { key := "ip:" + c.ClientIP() if v, ok := c.Get(CtxUserID); ok { @@ -81,8 +81,7 @@ func RateLimit(cache *store.Redis) gin.HandlerFunc { key = "u:" + uid } } - ok, _ := cache.Allow(c.Request.Context(), key, perMinute, time.Minute) - if !ok { + if !allowWithFallback(cache, c.Request.Context(), key, perMinute) { c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{"error": "rate limit exceeded"}) return } diff --git a/sundynix-gateway/internal/middleware/ratelimit.go b/sundynix-gateway/internal/middleware/ratelimit.go new file mode 100644 index 0000000..745cbe9 --- /dev/null +++ b/sundynix-gateway/internal/middleware/ratelimit.go @@ -0,0 +1,78 @@ +package middleware + +import ( + "context" + "net/http" + "sync" + "time" + + "github.com/gin-gonic/gin" + + "github.com/sundynix/sundynix-gateway/internal/store" +) + +// 限流的 fail-safe 兜底。Redis 是限流的主后端,但它一挂(或降级)时,此前 `Allow` 直接放行 +// = fail-open:爆破/洪泛防护随 Redis 一起消失。这里加一个**进程内**固定窗口兜底限流,让 +// Redis 不可用时仍有每实例的宽松限流(fail-safe),不牺牲整体可用性(本地限流很快、无外部依赖)。 + +// procLimiter 是进程内固定窗口计数器(每实例独立,不追求精确——兜底而已)。 +type procLimiter struct { + mu sync.Mutex + counts map[string]*winCount + window time.Duration +} + +type winCount struct { + n int + reset time.Time +} + +func newProcLimiter(window time.Duration) *procLimiter { + return &procLimiter{counts: make(map[string]*winCount), window: window} +} + +// allow 固定窗口内对 key 累加,超 limit 拒绝。顺带惰性清理过期项防 map 无界增长。 +func (l *procLimiter) allow(key string, limit int, now time.Time) bool { + l.mu.Lock() + defer l.mu.Unlock() + if len(l.counts) > 10000 { + for k, wc := range l.counts { + if now.After(wc.reset) { + delete(l.counts, k) + } + } + } + wc := l.counts[key] + if wc == nil || now.After(wc.reset) { + l.counts[key] = &winCount{n: 1, reset: now.Add(l.window)} + return true + } + wc.n++ + return wc.n <= limit +} + +// 全局兜底器(1 分钟窗口,与 Redis 限流同窗口口径)。 +var fallbackLimiter = newProcLimiter(time.Minute) + +// allowWithFallback 优先用 Redis 限流;Redis 降级/故障(!Enabled 或 Allow 出错)时回落进程内兜底。 +// 返回 true=放行。这是把 fail-open 改成 fail-safe 的关键接缝。 +func allowWithFallback(cache *store.Redis, ctx context.Context, key string, limit int) bool { + ok, err := cache.Allow(ctx, key, int64(limit), time.Minute) + if !cache.Enabled() || err != nil { + return fallbackLimiter.allow(key, limit, time.Now()) + } + return ok +} + +// RateLimitN 对某类端点施加更严的独立限流(按 IP + prefix 分桶),用于登录/注册等爆破面大的 +// 公开端点。与全局 RateLimit 叠加(两道桶都过才放行)。Redis 故障时走进程内兜底。 +func RateLimitN(cache *store.Redis, perMinute int, prefix string) gin.HandlerFunc { + return func(c *gin.Context) { + key := prefix + ":" + c.ClientIP() + if !allowWithFallback(cache, c.Request.Context(), key, perMinute) { + c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{"error": "请求过于频繁,请稍后再试"}) + return + } + c.Next() + } +} diff --git a/sundynix-gateway/internal/middleware/ratelimit_auth_test.go b/sundynix-gateway/internal/middleware/ratelimit_auth_test.go new file mode 100644 index 0000000..f920940 --- /dev/null +++ b/sundynix-gateway/internal/middleware/ratelimit_auth_test.go @@ -0,0 +1,56 @@ +package middleware + +import ( + "net/http/httptest" + "testing" + "time" + + "github.com/gin-gonic/gin" + + "github.com/sundynix/sundynix-gateway/internal/auth" +) + +// A5:进程内兜底限流——固定窗口内超限即拒,过窗后恢复。Redis 挂时靠它 fail-safe。 +func TestProcLimiter(t *testing.T) { + l := newProcLimiter(time.Minute) + now := time.Now() + for i := 0; i < 3; i++ { + if !l.allow("k", 3, now) { + t.Fatalf("前 3 次应放行(第 %d 次被拒)", i+1) + } + } + if l.allow("k", 3, now) { + t.Fatal("超限第 4 次应拒") + } + if !l.allow("k", 3, now.Add(2*time.Minute)) { + t.Fatal("新窗口应恢复放行") + } +} + +// A6:AuthFromHeaderOrQuery——?token= 有效则注入 uid、无 token 则 401。 +func TestAuthFromHeaderOrQuery(t *testing.T) { + gin.SetMode(gin.TestMode) + tok, err := auth.Issue("u1") + if err != nil { + t.Fatalf("签发失败: %v", err) + } + + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest("GET", "/x?token="+tok, nil) + AuthFromHeaderOrQuery()(c) + if c.IsAborted() { + t.Fatal("有效 query token 不该被拦") + } + if v, _ := c.Get(CtxUserID); v != "u1" { + t.Fatalf("应注入 uid=u1,得 %v", v) + } + + w2 := httptest.NewRecorder() + c2, _ := gin.CreateTestContext(w2) + c2.Request = httptest.NewRequest("GET", "/x", nil) + AuthFromHeaderOrQuery()(c2) + if !c2.IsAborted() || w2.Code != 401 { + t.Fatalf("无 token 应 401,得 aborted=%v code=%d", c2.IsAborted(), w2.Code) + } +} diff --git a/sundynix-gateway/internal/router/router.go b/sundynix-gateway/internal/router/router.go index 7597046..0d24ae5 100644 --- a/sundynix-gateway/internal/router/router.go +++ b/sundynix-gateway/internal/router/router.go @@ -59,15 +59,18 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob. // 微信扫码登录(全公开:ticket 是唯一凭证;PC 建票与轮询都无鉴权头) api.POST("/wx/mp/ticket", h.WxMPTicket) // PC 建票 + 微信二维码 api.GET("/wx/mp/poll", h.WxMPPoll) // PC 轮询登录态 - api.POST("/auth/register", h.Register) // 注册 + 签发 JWT - api.POST("/auth/login", h.Login) // 登录 + 签发 JWT + // 登录/注册爆破面大:在全局限流之上再叠一道更严的按 IP 专用限流(10/min)。 + api.POST("/auth/register", middleware.RateLimitN(cache, 10, "auth-reg"), h.Register) // 注册 + 签发 JWT + api.POST("/auth/login", middleware.RateLimitN(cache, 10, "auth-login"), h.Login) // 登录 + 签发 JWT api.GET("/auth/me", h.Me) // 当前登录用户(无效令牌 → 401) api.GET("/health", h.Health) // 依赖健康聚合(顶栏五盏灯) - api.GET("/tasks/:id/stream", h.StreamTask) // SSE 回流 Token Stream(task_id 寻址) - api.GET("/tasks/:id/exec", h.StreamExec) // SSE 回流执行轨迹(task_id 寻址) - api.GET("/kb/ingest/:id/stream", h.KbIngestStream) // 入库进度 SSE(job_id 寻址) - api.GET("/reports/:id/export", h.ExportReport) // 按需导出(report_id 寻址) - api.GET("/reports/:id/download", h.ExportReport) // 兼容旧入口(默认 docx) + // 这些 by-id 端点带不了 Bearer 头(EventSource/下载),改由 AuthFromHeaderOrQuery + // 从 ?token= 取 JWT 强制鉴权;task/report 三个再按 owner 归属校验(handler 内)。 + api.GET("/tasks/:id/stream", middleware.AuthFromHeaderOrQuery(), h.StreamTask) // SSE 回流 Token Stream + api.GET("/tasks/:id/exec", middleware.AuthFromHeaderOrQuery(), h.StreamExec) // SSE 回流执行轨迹 + api.GET("/kb/ingest/:id/stream", middleware.AuthFromHeaderOrQuery(), h.KbIngestStream) // 入库进度 SSE(登录即可,进度非敏感) + api.GET("/reports/:id/export", middleware.AuthFromHeaderOrQuery(), h.ExportReport) // 按需导出 + api.GET("/reports/:id/download", middleware.AuthFromHeaderOrQuery(), h.ExportReport) // 兼容旧入口(默认 docx) api.POST("/billing/callback/:channel", h.PaymentCallback) // 支付回调(渠道服务器带不了 Bearer;渠道验签是唯一的门) // —— 受保护:owner 作用域业务,必须携带有效 JWT —— @@ -75,7 +78,7 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob. { p.POST("/tasks", middleware.RequireTenantRole(db, store.RoleMember), h.SubmitTask) // 提交任务(烧租户积分):viewer 只读拦下 p.GET("/tasks/:id", h.TaskStatus) // 任务生命周期状态(UI 轮询 submitted/running/done/failed/timeout/waiting/rejected) - p.POST("/tasks/:id/approve", middleware.Audit(db), h.ApproveTask) // HITL 人工审批决定(批准/拒绝,审计) + p.POST("/tasks/:id/approve", middleware.RequireTenantRole(db, store.RoleMember), middleware.Audit(db), h.ApproveTask) // HITL 人工审批:≥member(放行会烧租户积分,viewer 拦下),审计 p.GET("/tenants/current", h.TenantCurrent) // 当前租户上下文 + 角色 + 可花余额(多租户) p.GET("/me/tenants", h.MyTenantsList) // 我所属租户(供切换) p.POST("/me/tenants", h.CreateMyTenant) // 自助建组织(创建者即 owner,建完切入) @@ -104,11 +107,13 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob. p.GET("/kb/doc", h.KbDoc) // 取单篇文档 p.DELETE("/kb/doc", middleware.RequireSpaceRole(db, store.RoleMember), h.KbDeleteDoc) // 级联删文档:viewer 拦下 - // Prompt 控制面(平台级配置:建版本 → 激活 → 控制面热下发各服务) - p.GET("/prompts", h.PromptList) // 列出全部版本 + 可配键 - p.POST("/prompts/version", h.PromptCreateVersion) // 建新版本(不自动激活) - p.POST("/prompts/activate", middleware.Audit(db), h.PromptActivate) // 激活某版本 → 广播热更新(审计) - p.POST("/prompts/deactivate", middleware.Audit(db), h.PromptDeactivate) // 撤销激活 → 回退代码默认(热,审计) + // Prompt 控制面(平台级配置:建版本 → 激活 → 控制面热下发各服务)。 + // 全平台配置、activate 热广播到所有服务、影响所有租户输出 —— 必须 RequireAdmin, + // 否则任意登录用户可改全局提示词。读写同源都挂 admin。 + p.GET("/prompts", middleware.RequireAdmin(), h.PromptList) // 列出全部版本 + 可配键 + p.POST("/prompts/version", middleware.RequireAdmin(), middleware.Audit(db), h.PromptCreateVersion) // 建新版本(不自动激活) + p.POST("/prompts/activate", middleware.RequireAdmin(), middleware.Audit(db), h.PromptActivate) // 激活某版本 → 广播热更新(审计) + p.POST("/prompts/deactivate", middleware.RequireAdmin(), middleware.Audit(db), h.PromptDeactivate) // 撤销激活 → 回退代码默认(热,审计) p.GET("/kb/links", h.KbLinks) // 某库双链 p.POST("/kb/note", middleware.RequireSpaceRole(db, store.RoleMember), h.KbSaveNote) // 新建/编辑笔记:viewer 拦下 p.GET("/kb/graph", h.KbGraph) // 知识图谱三元组 diff --git a/sundynix-gateway/internal/store/completeness_fixes_test.go b/sundynix-gateway/internal/store/completeness_fixes_test.go index 754cc3d..7a18df2 100644 --- a/sundynix-gateway/internal/store/completeness_fixes_test.go +++ b/sundynix-gateway/internal/store/completeness_fixes_test.go @@ -82,3 +82,19 @@ func TestListInvites_FiltersDeadCodes(t *testing.T) { t.Fatalf("全量列表应 3 条,得 %d", len(all)) } } + +// A6:TaskOwner 供公开 by-id 端点判权——按 task_id 返回提交者,跨租户可查,不存在返回空。 +func TestTaskOwner(t *testing.T) { + p := newTestStore(t) + seedTenant(t, p, "t1") + ctx := context.Background() + if err := p.SaveTask(ctx, "user-1", "task-abc", "{}"); err != nil { + t.Fatalf("建任务失败: %v", err) + } + if got := p.TaskOwner(ctx, "task-abc"); got != "user-1" { + t.Fatalf("owner 应 user-1,得 %q", got) + } + if got := p.TaskOwner(ctx, "nope"); got != "" { + t.Fatalf("不存在任务应返回空,得 %q", got) + } +} diff --git a/sundynix-gateway/internal/store/task_query.go b/sundynix-gateway/internal/store/task_query.go index 3bf63d5..e9262b5 100644 --- a/sundynix-gateway/internal/store/task_query.go +++ b/sundynix-gateway/internal/store/task_query.go @@ -98,6 +98,18 @@ type AdminEval struct { Corrected bool `json:"corrected"` } +// TaskOwner 按 task_id 返回提交者 user.id(供 SSE/导出等公开 by-id 端点做归属校验)。 +// 跨租户查(WithoutTenant):这些端点无租户上下文,靠 owner 判权。不存在返回空串。 +func (p *Postgres) TaskOwner(ctx context.Context, taskID string) string { + if p.db == nil || taskID == "" { + return "" + } + var owner string + p.db.WithContext(WithoutTenant(ctx)).Model(&Task{}). + Where("task_id = ? and deleted_at is null", taskID).Select("owner").Scan(&owner) + return owner +} + // TaskDetail 按 task_id 取单条任务的完整下钻数据(管理端,跨租户)。 // 必须 WithoutTenant:Task/Eval 都在租户插件作用域内,用请求 ctx 查别的租户的任务 // 不会报错,而是静默返回空——看起来像"这任务没产出",比报错更难排查。