diff --git a/cmd/gateway-maintenance/main.go b/cmd/gateway-maintenance/main.go index 9671b36..b2ba4e2 100644 --- a/cmd/gateway-maintenance/main.go +++ b/cmd/gateway-maintenance/main.go @@ -42,6 +42,12 @@ func main() { } else if runErr == nil { logger.Info("audit maintenance complete", "created_partitions", result.CreatedPartitions, "dropped_partitions", result.DroppedPartitions, "deleted_audit_rows", result.DeletedAuditRows, "deleted_usage_rows", result.DeletedUsageRows, "deleted_trace_rows", result.DeletedTraceRows) } + // 清理工具限流固定窗口:保留 2 天,防止 tool_rate_usage 无限膨胀。 + if tag, cleanupErr := db.Exec(ctx, `DELETE FROM gateway.tool_rate_usage WHERE window_start < clock_timestamp()-interval '2 days'`); cleanupErr != nil { + logger.Warn("tool rate window cleanup failed", "error", cleanupErr) + } else if tag.RowsAffected() > 0 { + logger.Info("tool rate windows cleaned", "rows", tag.RowsAffected()) + } select { case <-ctx.Done(): logger.Info("maintenance worker stopped") diff --git a/docs/security-review-0.10.1.md b/docs/security-review-0.10.1.md index eecd5eb..d515d34 100644 --- a/docs/security-review-0.10.1.md +++ b/docs/security-review-0.10.1.md @@ -511,3 +511,43 @@ PostgreSQL `text` 列拒绝写入 → 重试循环永远失败；inbox 的 Go `l least_conn 负载均衡,冷备含 master key 同步与恢复演练要点。 4. 实测:真实 Agent 进程注册→心跳上线→custom/http 任务认领执行上报→ 管理端成功可见。 + +# 追加：第七轮代码审查与修复（2026-08-13） + +三轮并行审查(安全/并发一致性/前端)后修复清单: + +**安全(H)** +- 渠道入站无鉴权:webhook 渠道强制 X-Inbound-Token 恒定时间校验,未配置令牌 + 拒绝入站;渠道不存在与令牌无效统一 401 文案防枚举;body 读取改 + io.ReadAll+LimitReader。企微签名按官方算法补 echostr 参与签名。 +- 运维报表/概览/systemInfo 端点由"仅认证"改为按权限 + (usage:read/audit:read/system:manage);reload 错误不再回显内部细节。 +- 扫码回调 sso_error 只回传固定码(unbound/disabled/expired/login_failed), + 内部细节写服务端日志。 +- 个人渠道令牌只接受请求头(query 会进日志/历史),渠道存在性不再可枚举。 +- 工具出站 SSRF:Dialer.Control 在系统 connect 阶段校验目标 IP,消除 + DNS rebinding 的校验-拨号 TOCTOU 窗口。 +- 新增 channel:read/channel:manage 权限并映射进 superadmin/auditor, + 渠道端点不再复用 notification 权限;个人限流倍数上限 100→10。 + +**并发/一致性(H/M)** +- 任务上报改为单条条件 UPDATE(claim_token+status 双守卫 + RowsAffected + 校验),并发/重放只有一个生效;认领支持回收过期 claimed 任务 + (available_at 即租约到期),节点崩溃不再永久卡死队列。 +- 资源申请审批改"先开通后落审批"两步式:Install/Grant 幂等且独立提交, + 不再事务内嵌套事务;渠道审批校验 enabled。 +- 聊天消息一轮(user+assistant)单事务落库,重试不产生孤儿/重复消息。 +- 会话列表校验 AuthVersion(改密/2FA 后旧会话不再展示);吊销先 Del 后 + SRem,Redis 抖动不再出现"列表已吊销但令牌仍有效"。 +- 删除工具前检查调用历史,有记录禁止删除(级联会清空报表数据)。 +- 工具审批 rejected 24h 冷却防通知轰炸;限流被拒调用补偿递减不占额度。 +- maintenance worker 每日清理 tool_rate_usage 过期窗口(保留 2 天)。 + +**前端/菜单(H/M)** +- 网关菜单 late-append bug:Reports/Tenants/Channels 因 gatewayChildren + cap 扩容指向旧数组而在菜单中不可见——已移到 push 前并加注释防复发。 +- 聊天会话改名 405:后端 PATCH 改 PUT 与前端 request.put 对齐。 +- 渠道编辑弹窗:openEdit 显式清空平台凭据字段(防跨渠道串写); + 新增启用开关,编辑不再静默重新启用停用渠道。 +- 聊天发送中切模型/会话:响应到达校验会话 id,丢弃迟到响应防串扰。 +- 报表默认日期改本地时区计算。 diff --git a/internal/agentnode/tasks.go b/internal/agentnode/tasks.go index 5f353f9..01b0d05 100644 --- a/internal/agentnode/tasks.go +++ b/internal/agentnode/tasks.go @@ -252,7 +252,8 @@ func (s *Store) ClaimTask(ctx context.Context, code, token string) (Task, error) err = s.pool.QueryRow(ctx, `UPDATE gateway.agent_tasks SET status='claimed',claim_token=$1,claimed_at=clock_timestamp(),available_at=clock_timestamp()+interval '15 minutes' WHERE id = ( SELECT t.id FROM gateway.agent_tasks t - WHERE t.status='queued' AND t.available_at<=clock_timestamp() + WHERE (t.status='queued' OR (t.status='claimed' AND t.available_at<=clock_timestamp())) + AND t.available_at<=clock_timestamp() AND (t.node_id IS NULL OR t.node_id=$2) AND (t.node_id IS NOT NULL OR (t.pool_type=$3 AND t.pool_code=$4)) ORDER BY t.created_at LIMIT 1 FOR UPDATE SKIP LOCKED @@ -291,18 +292,16 @@ func (s *Store) CompleteTask(ctx context.Context, code, token, taskID, claimToke return Task{}, fmt.Errorf("%w: %v", ErrStore, err) } defer func() { _ = tx.Rollback(ctx) }() - var status, currentError string + // 单条条件 UPDATE 完成状态机转移:claim_token + status 双守卫保证 + // 并发/重放上报只有一个能生效(RowsAffected==1),不会出现读-改-写窗口。 var attempts, maxAttempts int - err = tx.QueryRow(ctx, `SELECT status,error,attempts,max_attempts FROM gateway.agent_tasks WHERE id=$1 AND claim_token=$2`, taskID, claimToken).Scan(&status, ¤tError, &attempts, &maxAttempts) + err = tx.QueryRow(ctx, `SELECT attempts,max_attempts FROM gateway.agent_tasks WHERE id=$1::uuid AND claim_token=$2 AND status IN ('claimed','running') FOR UPDATE`, taskID, claimToken).Scan(&attempts, &maxAttempts) if errors.Is(err, pgx.ErrNoRows) { return Task{}, ErrTaskConflict } if err != nil { return Task{}, fmt.Errorf("%w: %v", ErrStore, err) } - if status != "claimed" && status != "running" { - return Task{}, ErrTaskConflict - } attempts++ nextStatus := "succeeded" if taskError != "" { @@ -313,11 +312,14 @@ func (s *Store) CompleteTask(ctx context.Context, code, token, taskID, claimToke } // 失败退避:30s * 已尝试次数。 backoff := 30 * time.Second * time.Duration(attempts) - _, err = tx.Exec(ctx, `UPDATE gateway.agent_tasks SET status=$2::text,attempts=$3::int,result=$4::jsonb,error=$5::text,claim_token=NULL,finished_at=CASE WHEN $2::text IN ('succeeded','failed','cancelled') THEN clock_timestamp() ELSE NULL END,available_at=CASE WHEN $2::text='queued' THEN clock_timestamp()+$6::interval ELSE available_at END WHERE id=$1::uuid`, - taskID, nextStatus, attempts, normalizeJSON(result), taskError, backoff) + tag, err := tx.Exec(ctx, `UPDATE gateway.agent_tasks SET status=$2::text,attempts=$3::int,result=$4::jsonb,error=$5::text,claim_token=NULL,finished_at=CASE WHEN $2::text IN ('succeeded','failed','cancelled') THEN clock_timestamp() ELSE NULL END,available_at=CASE WHEN $2::text='queued' THEN clock_timestamp()+$6::interval ELSE available_at END WHERE id=$1::uuid AND claim_token=$7 AND status IN ('claimed','running')`, + taskID, nextStatus, attempts, normalizeJSON(result), taskError, backoff, claimToken) if err != nil { return Task{}, fmt.Errorf("%w: %v", ErrStore, err) } + if tag.RowsAffected() != 1 { + return Task{}, ErrTaskConflict + } eventID, _ := platformid.NewUUID() eventType := "agent_task.completed" if nextStatus == "failed" { diff --git a/internal/channel/http.go b/internal/channel/http.go index 9628b9a..624407c 100644 --- a/internal/channel/http.go +++ b/internal/channel/http.go @@ -2,7 +2,9 @@ package channel import ( "context" + "crypto/subtle" "encoding/json" + "io" "net/http" "strings" @@ -46,7 +48,7 @@ func (h *HTTPHandler) require(w http.ResponseWriter, r *http.Request, permission } func (h *HTTPHandler) list(w http.ResponseWriter, r *http.Request) { - if _, ok := h.require(w, r, identity.PermissionNotificationRead); !ok { + if _, ok := h.require(w, r, identity.PermissionChannelRead); !ok { return } items, err := h.service.List(r.Context()) @@ -69,7 +71,7 @@ type channelInput struct { } func (h *HTTPHandler) save(w http.ResponseWriter, r *http.Request) { - actor, ok := h.require(w, r, identity.PermissionNotificationManage) + actor, ok := h.require(w, r, identity.PermissionChannelManage) if !ok { return } @@ -100,7 +102,7 @@ func (h *HTTPHandler) save(w http.ResponseWriter, r *http.Request) { } func (h *HTTPHandler) delete(w http.ResponseWriter, r *http.Request) { - if _, ok := h.require(w, r, identity.PermissionNotificationManage); !ok { + if _, ok := h.require(w, r, identity.PermissionChannelManage); !ok { return } if err := h.service.Delete(r.Context(), r.PathValue("id")); err != nil { @@ -112,7 +114,7 @@ func (h *HTTPHandler) delete(w http.ResponseWriter, r *http.Request) { // test 用渠道绑定模型发送一条测试消息并尝试平台回复。 func (h *HTTPHandler) test(w http.ResponseWriter, r *http.Request) { - if _, ok := h.require(w, r, identity.PermissionNotificationManage); !ok { + if _, ok := h.require(w, r, identity.PermissionChannelManage); !ok { return } items, err := h.service.List(r.Context()) @@ -160,7 +162,8 @@ func (h *InboundHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (h *InboundHTTPHandler) inbound(w http.ResponseWriter, r *http.Request) { c, err := h.service.GetByCode(r.Context(), strings.ToLower(r.PathValue("code"))) if err != nil { - apiresponse.Error(w, http.StatusNotFound, "渠道不存在或未启用") + // 与令牌无效返回同一错误,不泄露渠道存在性。 + apiresponse.Error(w, http.StatusUnauthorized, "渠道令牌无效或渠道不存在") return } cfg, err := h.service.DecryptConfig(c) @@ -180,8 +183,19 @@ func (h *InboundHTTPHandler) inbound(w http.ResponseWriter, r *http.Request) { apiresponse.Error(w, http.StatusUnauthorized, "签名校验失败") return } - case "dingtalk": - // 钉钉机器人验签由平台侧 access_token 控制;此处信任令牌。 + case "dingtalk", "feishu": + // 钉钉/飞书机器人验签由平台侧 access_token/回调令牌控制;此处信任平台。 + case "webhook": + // 通用 webhook 必须携带入站令牌:未配置令牌的渠道拒绝入站,防止 + // 任意调用者消耗绑定模型的配额与费用。 + presented := strings.TrimSpace(r.Header.Get("X-Inbound-Token")) + if presented == "" { + presented = strings.TrimSpace(r.URL.Query().Get("token")) + } + if cfg.InboundToken == "" || subtle.ConstantTimeCompare([]byte(presented), []byte(cfg.InboundToken)) != 1 { + apiresponse.Error(w, http.StatusUnauthorized, "渠道令牌无效或渠道不存在") + return + } } var payload struct { Text struct { @@ -189,9 +203,8 @@ func (h *InboundHTTPHandler) inbound(w http.ResponseWriter, r *http.Request) { } `json:"text"` Content string `json:"content"` } - raw := make([]byte, 1<<20) - n, _ := r.Body.Read(raw) - _ = json.Unmarshal(raw[:n], &payload) + raw, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20)) + _ = json.Unmarshal(raw, &payload) text := payload.Text.Content if text == "" { text = payload.Content @@ -219,7 +232,7 @@ func (h *InboundHTTPHandler) inbound(w http.ResponseWriter, r *http.Request) { // listGrants 渠道用户授权列表。 func (h *HTTPHandler) listGrants(w http.ResponseWriter, r *http.Request) { - if _, ok := h.require(w, r, identity.PermissionNotificationRead); !ok { + if _, ok := h.require(w, r, identity.PermissionChannelRead); !ok { return } items, err := h.service.ListGrants(r.Context(), r.PathValue("id")) @@ -232,7 +245,7 @@ func (h *HTTPHandler) listGrants(w http.ResponseWriter, r *http.Request) { // grant 直接授予用户渠道使用权限(管理员显式授权,无需走申请流)。 func (h *HTTPHandler) grant(w http.ResponseWriter, r *http.Request) { - actor, ok := h.require(w, r, identity.PermissionNotificationManage) + actor, ok := h.require(w, r, identity.PermissionChannelManage) if !ok { return } @@ -254,7 +267,7 @@ func (h *HTTPHandler) grant(w http.ResponseWriter, r *http.Request) { // revokeGrant 撤销用户的渠道使用权限。 func (h *HTTPHandler) revokeGrant(w http.ResponseWriter, r *http.Request) { - if _, ok := h.require(w, r, identity.PermissionNotificationManage); !ok { + if _, ok := h.require(w, r, identity.PermissionChannelManage); !ok { return } if err := h.service.RevokeGrant(r.Context(), r.PathValue("id"), r.PathValue("user_id")); err != nil { diff --git a/internal/channel/service.go b/internal/channel/service.go index 8cb72d5..ef3b514 100644 --- a/internal/channel/service.go +++ b/internal/channel/service.go @@ -183,6 +183,9 @@ func (s *Service) Save(ctx context.Context, id, code, name, kind string, cfg Con if modelBinding == nil { modelBinding = json.RawMessage(`{}`) } + if departmentIDs == nil { + departmentIDs = []string{} + } _, err = s.pool.Exec(ctx, `INSERT INTO gateway.channels(id,code,name,kind,encrypted_config,config_kek_version,encrypted_api_key,api_key_kek_version,model_binding,department_ids,enabled,created_by) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT(code) DO UPDATE SET name=$3,kind=$4,encrypted_config=$5,config_kek_version=$6, @@ -415,9 +418,12 @@ func (s *Service) postJSON(ctx context.Context, endpoint string, payload []byte, return nil } -// VerifyWeComSignature 校验企业微信回调签名(URL 参数签名)。 +// VerifyWeComSignature 校验企业微信回调签名。 +// 官方算法:token、timestamp、nonce、echostr(或加密消息)四个参数按字典序 +// 排序后拼接,取 SHA1 十六进制与 msg_signature 比较。values 用于带密文消息 +// 体校验时补充参与签名计算的参数。 func VerifyWeComSignature(token, timestamp, nonce, echostr string, values map[string]string) (string, bool) { - parts := []string{token, timestamp, nonce} + parts := []string{token, timestamp, nonce, echostr} if values != nil { keys := make([]string, 0, len(values)) for key := range values { diff --git a/internal/identity/account.go b/internal/identity/account.go index f667041..dcb1cbf 100644 --- a/internal/identity/account.go +++ b/internal/identity/account.go @@ -65,6 +65,8 @@ const ( PermissionApplicationManage = "application:manage" PermissionNotificationRead = "notification:read" PermissionNotificationManage = "notification:manage" + PermissionChannelRead = "channel:read" + PermissionChannelManage = "channel:manage" PermissionMCPServerRead = "mcp_server:read" PermissionMCPServerManage = "mcp_server:manage" PermissionSkillRead = "skill:read" @@ -99,6 +101,7 @@ var rolePermissions = map[string][]string{ PermissionToolRead, PermissionToolManage, PermissionApplicationRead, PermissionApplicationManage, PermissionNotificationRead, PermissionNotificationManage, + PermissionChannelRead, PermissionChannelManage, PermissionMCPServerRead, PermissionMCPServerManage, PermissionSkillRead, PermissionSkillManage, PermissionDigitalEmployeeRead, PermissionDigitalEmployeeManage, @@ -110,7 +113,7 @@ var rolePermissions = map[string][]string{ PermissionAgentNodeRead, PermissionAgentNodeManage, PermissionSystemManage, }, - "auditor": {PermissionProviderRead, PermissionAPIKeyRead, PermissionAuditRead, PermissionUsageRead, PermissionOutboxRead, PermissionContentPolicyRead, PermissionPricingRead, PermissionPromptRead, PermissionKnowledgeRead, PermissionToolRead, PermissionApplicationRead, PermissionNotificationRead, PermissionMCPServerRead, PermissionSkillRead, PermissionDigitalEmployeeRead, PermissionMarketplaceRead, PermissionFileRead, PermissionInboxRead, PermissionScheduledTaskRead, PermissionTraceRead, PermissionAgentNodeRead}, + "auditor": {PermissionProviderRead, PermissionAPIKeyRead, PermissionAuditRead, PermissionUsageRead, PermissionOutboxRead, PermissionContentPolicyRead, PermissionPricingRead, PermissionPromptRead, PermissionKnowledgeRead, PermissionToolRead, PermissionApplicationRead, PermissionNotificationRead, PermissionChannelRead, PermissionMCPServerRead, PermissionSkillRead, PermissionDigitalEmployeeRead, PermissionMarketplaceRead, PermissionFileRead, PermissionInboxRead, PermissionScheduledTaskRead, PermissionTraceRead, PermissionAgentNodeRead}, "member": {}, } diff --git a/internal/identity/http.go b/internal/identity/http.go index e6b0f47..8d5d25e 100644 --- a/internal/identity/http.go +++ b/internal/identity/http.go @@ -468,6 +468,16 @@ func adminMenus(account Account) []map[string]any { if HasPermission(account, PermissionPricingRead) || HasPermission(account, PermissionPricingManage) { gatewayChildren = append(gatewayChildren, map[string]any{"name": "ModelPrices", "path": "model-prices", "component": "/gateway/model-prices", "meta": map[string]any{"title": "模型价格"}}) } + if HasPermission(account, PermissionUsageRead) { + gatewayChildren = append(gatewayChildren, map[string]any{"name": "Reports", "path": "reports", "component": "/gateway/reports", "meta": map[string]any{"title": "企业报表"}}) + gatewayChildren = append(gatewayChildren, map[string]any{"name": "Tenants", "path": "tenants", "component": "/gateway/tenants", "meta": map[string]any{"title": "租户概览"}}) + } + if HasPermission(account, PermissionNotificationRead) || HasPermission(account, PermissionNotificationManage) { + gatewayChildren = append(gatewayChildren, map[string]any{"name": "Channels", "path": "channels", "component": "/gateway/channels", "meta": map[string]any{"title": "渠道管理"}}) + } + // 注意:所有子菜单必须在 push 进 menus 前 append 完——gatewayChildren 容量 + // 有限,先 push 后 append 会因扩容导致 menus 里的 children 指向旧数组, + // 后加页面在菜单中不可见。 if len(gatewayChildren) > 0 { menus = append(menus, map[string]any{"name": "Gateway", "path": "/gateway", "component": "/index/index", "meta": map[string]any{"title": "网关接入", "icon": "ri:router-line"}, "children": gatewayChildren}) } @@ -534,13 +544,6 @@ func adminMenus(account Account) []map[string]any { menus = append(menus, map[string]any{"name": "ResourceMarket", "path": "/resource-market", "component": "/index/index", "meta": map[string]any{"title": "资源市场", "icon": "ri:store-3-line"}, "children": marketChildren}) } - if HasPermission(account, PermissionUsageRead) { - gatewayChildren = append(gatewayChildren, map[string]any{"name": "Reports", "path": "reports", "component": "/gateway/reports", "meta": map[string]any{"title": "企业报表"}}) - gatewayChildren = append(gatewayChildren, map[string]any{"name": "Tenants", "path": "tenants", "component": "/gateway/tenants", "meta": map[string]any{"title": "租户概览"}}) - } - if HasPermission(account, PermissionNotificationRead) || HasPermission(account, PermissionNotificationManage) { - gatewayChildren = append(gatewayChildren, map[string]any{"name": "Channels", "path": "channels", "component": "/gateway/channels", "meta": map[string]any{"title": "渠道管理"}}) - } // 系统管理：账号权限、事件投递与通知。 systemChildren := make([]map[string]any, 0, 3) if HasPermission(account, PermissionIdentityManage) { diff --git a/internal/identity/session.go b/internal/identity/session.go index 4d60515..15c8320 100644 --- a/internal/identity/session.go +++ b/internal/identity/session.go @@ -248,6 +248,11 @@ func (s *SessionStore) ListSessions(ctx context.Context, kind Kind, subjectID, c _ = s.client.SRem(ctx, sessionIndexKey(kind, subjectID), hex).Err() continue } + // 凭据版本不匹配的会话(改密/2FA 变更后)实际已失效,不展示并清理索引。 + if principal.AuthVersion != s.AuthVersion(ctx, principal.Kind, principal.SubjectID) { + _ = s.client.SRem(ctx, sessionIndexKey(kind, subjectID), hex).Err() + continue + } items = append(items, SessionView{ID: hex, IP: principal.IP, UserAgent: principal.UserAgent, IssuedAt: principal.IssuedAt, Current: s.sessionHexMatches(currentAuthorization, hex)}) } @@ -269,6 +274,11 @@ func (s *SessionStore) RevokeSession(ctx context.Context, kind Kind, subjectID, if s.sessionHexMatches(currentAuthorization, sessionID) { return ErrRevokeCurrentSession } + if err := s.client.Del(ctx, sessionKeyFromHex(sessionID)).Err(); err != nil { + return fmt.Errorf("%w: %v", ErrUnavailable, err) + } + // 先删会话键再删索引:Del 失败时两者都保留(列表仍显示,可重试); + // SRem 失败只留脏索引,由 ListSessions 惰性清理自愈。 removed, err := s.client.SRem(ctx, sessionIndexKey(kind, subjectID), sessionID).Result() if err != nil { return fmt.Errorf("%w: %v", ErrUnavailable, err) @@ -276,9 +286,6 @@ func (s *SessionStore) RevokeSession(ctx context.Context, kind Kind, subjectID, if removed == 0 { return ErrInvalidSession } - if err := s.client.Del(ctx, sessionKeyFromHex(sessionID)).Err(); err != nil { - return fmt.Errorf("%w: %v", ErrUnavailable, err) - } return nil } diff --git a/internal/identity/social_http.go b/internal/identity/social_http.go index 8a8afc3..5a15b14 100644 --- a/internal/identity/social_http.go +++ b/internal/identity/social_http.go @@ -1,6 +1,7 @@ package identity import ( + "errors" "net/http" "net/url" "strings" @@ -52,7 +53,18 @@ func (h *HTTPHandler) callbackSocial(w http.ResponseWriter, r *http.Request) { } result, err := h.service.CompleteSocialLogin(r.Context(), provider.Kind, state, code, SessionMeta{IP: h.service.ClientIP(r), UserAgent: r.UserAgent()}) if err != nil { - h.socialRedirect(w, r, provider, "sso_error", err.Error()) + // 只回传固定错误码,内部细节写服务端日志,避免内部信息进浏览器 + // 地址栏/历史/Referer。 + code := "login_failed" + switch { + case errors.Is(err, ErrSocialUnbound): + code = "unbound" + case errors.Is(err, ErrAccountDisabled): + code = "disabled" + case errors.Is(err, ErrInvalidSession): + code = "expired" + } + h.socialRedirect(w, r, provider, "sso_error", code) return } switch result.Purpose { diff --git a/internal/operations/admin_http.go b/internal/operations/admin_http.go index 6f7c49f..02ae118 100644 --- a/internal/operations/admin_http.go +++ b/internal/operations/admin_http.go @@ -43,8 +43,20 @@ func (h *AdminHTTPHandler) account(w http.ResponseWriter, r *http.Request) (iden return account, true } +func (h *AdminHTTPHandler) requirePermission(w http.ResponseWriter, r *http.Request, permission string) (identity.Account, bool) { + account, ok := h.account(w, r) + if !ok { + return account, false + } + if !identity.HasPermission(account, permission) { + apiresponse.Error(w, http.StatusForbidden, "缺少权限") + return identity.Account{}, false + } + return account, true +} + func (h *AdminHTTPHandler) systemInfo(w http.ResponseWriter, r *http.Request) { - if _, ok := h.account(w, r); !ok { + if _, ok := h.requirePermission(w, r, identity.PermissionSystemManage); !ok { return } var providers, models, keys int64 @@ -56,7 +68,7 @@ func (h *AdminHTTPHandler) systemInfo(w http.ResponseWriter, r *http.Request) { } func (h *AdminHTTPHandler) overview(w http.ResponseWriter, r *http.Request) { - if _, ok := h.account(w, r); !ok { + if _, ok := h.requirePermission(w, r, identity.PermissionUsageRead); !ok { return } var requests, failures, promptTokens, completionTokens, cost int64 @@ -80,7 +92,7 @@ func (h *AdminHTTPHandler) reloadSnapshots(w http.ResponseWriter, r *http.Reques } if h.reload != nil { if err := h.reload(r.Context()); err != nil { - apiresponse.Error(w, http.StatusServiceUnavailable, "运行时快照刷新不完整: "+err.Error()) + apiresponse.Error(w, http.StatusServiceUnavailable, "运行时快照刷新不完整，请查看服务端日志") return } } @@ -89,7 +101,7 @@ func (h *AdminHTTPHandler) reloadSnapshots(w http.ResponseWriter, r *http.Reques // tenantsOverview 以部门为租户维度,汇总各租户的账号/Key/用量与配额。 func (h *AdminHTTPHandler) tenantsOverview(w http.ResponseWriter, r *http.Request) { - if _, ok := h.account(w, r); !ok { + if _, ok := h.requirePermission(w, r, identity.PermissionUsageRead); !ok { return } rows, err := h.pool.Query(r.Context(), `SELECT d.id::text,d.name,d.max_api_keys,d.max_monthly_tokens, @@ -151,7 +163,7 @@ func (h *AdminHTTPHandler) reportRange(r *http.Request) (from, to time.Time) { // reportTools 工具维度统计:调用数/成功率/平均延迟。 func (h *AdminHTTPHandler) reportTools(w http.ResponseWriter, r *http.Request) { - if _, ok := h.account(w, r); !ok { + if _, ok := h.requirePermission(w, r, identity.PermissionUsageRead); !ok { return } from, to := h.reportRange(r) @@ -182,7 +194,7 @@ func (h *AdminHTTPHandler) reportTools(w http.ResponseWriter, r *http.Request) { // reportApprovals 审批维度统计:模型/资源/工具申请的发起与审批结果。 func (h *AdminHTTPHandler) reportApprovals(w http.ResponseWriter, r *http.Request) { - if _, ok := h.account(w, r); !ok { + if _, ok := h.requirePermission(w, r, identity.PermissionUsageRead); !ok { return } from, to := h.reportRange(r) @@ -209,7 +221,7 @@ func (h *AdminHTTPHandler) reportApprovals(w http.ResponseWriter, r *http.Reques // reportSecurity 安全维度统计:登录成功/失败、锁定与来源 IP 分布。 func (h *AdminHTTPHandler) reportSecurity(w http.ResponseWriter, r *http.Request) { - if _, ok := h.account(w, r); !ok { + if _, ok := h.requirePermission(w, r, identity.PermissionAuditRead); !ok { return } from, to := h.reportRange(r) diff --git a/internal/portal/chat.go b/internal/portal/chat.go index 2e58e34..93bd354 100644 --- a/internal/portal/chat.go +++ b/internal/portal/chat.go @@ -185,43 +185,55 @@ func (s *Service) ChatSession(ctx context.Context, account identity.Account, id return item, rows.Err() } -// appendChatMessage 在会话上追加一条消息(哈希链 + 序号,事务内完成)。 -func (s *Service) appendChatMessage(ctx context.Context, sessionID, role, content string) (ConversationMessage, error) { +// appendChatMessages 在会话上批量追加消息(user+assistant 一轮):单事务内 +// 连续插入、序列号一次锁定一次递增,模型调用成功后才落库——要么整轮落库 +// 要么整轮不落,客户端重试不会产生孤儿或重复消息。 +func (s *Service) appendChatMessages(ctx context.Context, sessionID string, messages []ConversationMessage) ([]ConversationMessage, error) { + if len(messages) == 0 { + return nil, nil + } tx, err := s.pool.Begin(ctx) if err != nil { - return ConversationMessage{}, err + return nil, err } defer func() { _ = tx.Rollback(ctx) }() var sequence int if err = tx.QueryRow(ctx, `SELECT next_sequence FROM gateway.portal_chat_sessions WHERE id=$1 FOR UPDATE`, sessionID).Scan(&sequence); err != nil { - return ConversationMessage{}, err + return nil, err } - if sequence > 200 { - return ConversationMessage{}, errors.New("本会话已达到 200 条消息上限") + if sequence+len(messages)-1 > 200 { + return nil, errors.New("本会话已达到 200 条消息上限") } previous := strings.Repeat("0", 64) if sequence > 1 { if err = tx.QueryRow(ctx, `SELECT message_hash FROM gateway.portal_chat_messages WHERE session_id=$1 AND sequence=$2`, sessionID, sequence-1).Scan(&previous); err != nil { - return ConversationMessage{}, err + return nil, err } } - id, err := platformid.NewUUID() - if err != nil { - return ConversationMessage{}, err + firstTitle := messages[0].Content + out := make([]ConversationMessage, 0, len(messages)) + for _, message := range messages { + id, err := platformid.NewUUID() + if err != nil { + return nil, err + } + hash := messageDigest(previous, sequence, message.Role, message.Content) + var created time.Time + if err = tx.QueryRow(ctx, `INSERT INTO gateway.portal_chat_messages(id,session_id,sequence,role,content,previous_hash,message_hash) VALUES($1,$2,$3,$4,$5,$6,$7) RETURNING created_at`, id, sessionID, sequence, message.Role, message.Content, previous, hash).Scan(&created); err != nil { + return nil, err + } + out = append(out, ConversationMessage{Sequence: sequence, Role: message.Role, Content: message.Content, CreatedAt: created}) + previous = hash + sequence++ } - hash := messageDigest(previous, sequence, role, content) - var created time.Time - if err = tx.QueryRow(ctx, `INSERT INTO gateway.portal_chat_messages(id,session_id,sequence,role,content,previous_hash,message_hash) VALUES($1,$2,$3,$4,$5,$6,$7) RETURNING created_at`, id, sessionID, sequence, role, content, previous, hash).Scan(&created); err != nil { - return ConversationMessage{}, err - } - _, err = tx.Exec(ctx, `UPDATE gateway.portal_chat_sessions SET next_sequence=next_sequence+1,title=CASE WHEN next_sequence=1 THEN left($2,160) ELSE title END,updated_at=clock_timestamp() WHERE id=$1`, sessionID, content) + _, err = tx.Exec(ctx, `UPDATE gateway.portal_chat_sessions SET next_sequence=$2,title=CASE WHEN next_sequence=1 THEN left($3,160) ELSE title END,updated_at=clock_timestamp() WHERE id=$1`, sessionID, sequence, firstTitle) if err != nil { - return ConversationMessage{}, err + return nil, err } if err = tx.Commit(ctx); err != nil { - return ConversationMessage{}, err + return nil, err } - return ConversationMessage{Sequence: sequence, Role: role, Content: content, CreatedAt: created}, nil + return out, nil } // callChat 用用户的运行时凭据直接调用受管网关 /v1/chat/completions。 @@ -317,10 +329,7 @@ func (s *Service) AppendChatMessage(ctx context.Context, account identity.Accoun if err != nil { return response, err } - if _, err = s.appendChatMessage(ctx, id, "user", message); err != nil { - return nil, err - } - if _, err = s.appendChatMessage(ctx, id, "assistant", answer); err != nil { + if _, err = s.appendChatMessages(ctx, id, []ConversationMessage{{Role: "user", Content: message}, {Role: "assistant", Content: answer}}); err != nil { return nil, err } response["conversation_id"] = id diff --git a/internal/portal/http.go b/internal/portal/http.go index b4aaba6..c5e22c2 100644 --- a/internal/portal/http.go +++ b/internal/portal/http.go @@ -53,7 +53,7 @@ func NewHTTPHandler(service *Service, identityService *identity.Service) *HTTPHa h.mux.HandleFunc("POST /api/v1/portal/chat/completions", h.chatOnce) h.mux.HandleFunc("GET /api/v1/portal/chat/sessions", h.listChatSessions) h.mux.HandleFunc("POST /api/v1/portal/chat/sessions", h.createChatSession) - h.mux.HandleFunc("PATCH /api/v1/portal/chat/sessions/{id}", h.renameChatSession) + h.mux.HandleFunc("PUT /api/v1/portal/chat/sessions/{id}", h.renameChatSession) h.mux.HandleFunc("DELETE /api/v1/portal/chat/sessions/{id}", h.deleteChatSession) h.mux.HandleFunc("GET /api/v1/portal/chat/sessions/{id}", h.getChatSession) h.mux.HandleFunc("POST /api/v1/portal/chat/sessions/{id}/messages", h.appendChatMessage) @@ -857,10 +857,8 @@ func (h *HTTPHandler) deletePersonalChannel(w http.ResponseWriter, r *http.Reque // personalChannelInbound 个人渠道入站(公开端点,令牌鉴权,同步返回文本)。 func (h *HTTPHandler) personalChannelInbound(w http.ResponseWriter, r *http.Request) { + // 令牌只经请求头传递:query 传参会进访问日志/浏览器历史/Referer。 token := strings.TrimSpace(r.Header.Get("X-Inbound-Token")) - if token == "" { - token = strings.TrimSpace(r.URL.Query().Get("token")) - } var input struct { Message string `json:"message"` Content string `json:"content"` diff --git a/internal/portal/personal_channels.go b/internal/portal/personal_channels.go index 2b5e28f..8c1cc05 100644 --- a/internal/portal/personal_channels.go +++ b/internal/portal/personal_channels.go @@ -139,7 +139,8 @@ func (s *Service) HandlePersonalInbound(ctx context.Context, code, presentedToke var id, userID, providerCode, model, tokenHash string err := s.pool.QueryRow(ctx, `SELECT id::text,portal_user_id::text,provider_code,model,inbound_token_hash FROM gateway.personal_channels WHERE code=$1 AND enabled`, code).Scan(&id, &userID, &providerCode, &model, &tokenHash) if errors.Is(err, pgx.ErrNoRows) { - return "", errors.New("渠道不存在或未启用") + // 与令牌无效同一文案,不泄露渠道存在性。 + return "", errors.New("入站令牌无效") } if err != nil { return "", err diff --git a/internal/portal/requests.go b/internal/portal/requests.go index 4aede22..e319aa6 100644 --- a/internal/portal/requests.go +++ b/internal/portal/requests.go @@ -155,7 +155,13 @@ func (s *Service) AdminResourceRequests(ctx context.Context, status string) ([]R return items, rows.Err() } -// DecideResourceRequest 审批资源/渠道申请:通过时自动开通(marketplace 安装)。 +// DecideResourceRequest 审批资源/渠道申请:通过时自动开通(marketplace 安装 / +// channel_grants 授权)。 +// +// 顺序为先开通、后落审批记录:开通方法(Install/Grant)是幂等的且各自独立 +// 提交,审批状态更新在同事务内与 outbox 事件一起提交。开通失败时申请保持 +// pending,管理员可重试,不会出现"记录已通过但未开通"或"事务内嵌套事务" +// 的中间态;开通成功但状态提交失败时,重试会幂等收敛。 func (s *Service) DecideResourceRequest(ctx context.Context, id, status, note, actorID string) (ResourceRequest, error) { if status != "approved" && status != "rejected" { return ResourceRequest{}, errors.New("审批状态无效") @@ -163,6 +169,40 @@ func (s *Service) DecideResourceRequest(ctx context.Context, id, status, note, a if len(note) > 4000 { return ResourceRequest{}, errors.New("审批备注过长") } + var userID, resourceType, resourceCode string + err := s.pool.QueryRow(ctx, `SELECT portal_user_id::text,resource_type,resource_code FROM gateway.resource_access_requests WHERE id=$1 AND status='pending'`, id).Scan(&userID, &resourceType, &resourceCode) + if errors.Is(err, pgx.ErrNoRows) { + return ResourceRequest{}, ErrNotFound + } + if err != nil { + return ResourceRequest{}, err + } + if status == "approved" { + switch resourceType { + case "mcp_server", "skill", "digital_employee": + if s.market != nil { + // 自动安装到申请用户工作区(use 等级,幂等)。 + if _, err = s.market.Install(ctx, resourceType, resourceCode, userID, "use"); err != nil { + return ResourceRequest{}, err + } + } + case "channel": + // 渠道审批通过 = 写入 channel_grants 用户级授权(幂等)。 + if s.channels != nil { + var channelID string + var enabled bool + if err = s.pool.QueryRow(ctx, `SELECT id::text,enabled FROM gateway.channels WHERE code=$1`, resourceCode).Scan(&channelID, &enabled); err != nil { + return ResourceRequest{}, errors.New("渠道不存在，无法开通") + } + if !enabled { + return ResourceRequest{}, errors.New("渠道已停用，无法开通") + } + if err = s.channels.Grant(ctx, channelID, userID, actorID, "approval"); err != nil { + return ResourceRequest{}, err + } + } + } + } eventID, _ := platformid.NewUUID() tx, err := s.pool.Begin(ctx) if err != nil { @@ -174,34 +214,9 @@ func (s *Service) DecideResourceRequest(ctx context.Context, id, status, note, a return ResourceRequest{}, err } if tag.RowsAffected() == 0 { + // 并发审批:后到者失败,开通动作已幂等,无副作用残留。 return ResourceRequest{}, ErrNotFound } - var userID, resourceType, resourceCode string - if err = tx.QueryRow(ctx, `SELECT portal_user_id::text,resource_type,resource_code FROM gateway.resource_access_requests WHERE id=$1`, id).Scan(&userID, &resourceType, &resourceCode); err != nil { - return ResourceRequest{}, err - } - if status == "approved" { - switch resourceType { - case "mcp_server", "skill", "digital_employee": - if s.market != nil { - // 自动安装到申请用户工作区(use 等级)。 - if _, err = s.market.Install(ctx, resourceType, resourceCode, userID, "use"); err != nil { - return ResourceRequest{}, err - } - } - case "channel": - // 渠道审批通过 = 写入 channel_grants 用户级授权。 - if s.channels != nil { - var channelID string - if err = tx.QueryRow(ctx, `SELECT id::text FROM gateway.channels WHERE code=$1`, resourceCode).Scan(&channelID); err != nil { - return ResourceRequest{}, err - } - if err = s.channels.Grant(ctx, channelID, userID, actorID, "approval"); err != nil { - return ResourceRequest{}, err - } - } - } - } payload, _ := json.Marshal(map[string]any{"request_id": id, "portal_user_id": userID, "resource_type": resourceType, "resource_code": resourceCode, "status": status, "actor_id": actorID}) if _, err = tx.Exec(ctx, `INSERT INTO gateway.outbox_events(event_id,event_type,event_version,aggregate_type,aggregate_id,payload) VALUES($1,'resource_access.decided',1,'resource_access_request',$2,$3)`, eventID, id, payload); err != nil { return ResourceRequest{}, err diff --git a/internal/workbench/agentpolicy.go b/internal/workbench/agentpolicy.go index f8492d8..683f932 100644 --- a/internal/workbench/agentpolicy.go +++ b/internal/workbench/agentpolicy.go @@ -50,8 +50,8 @@ func (s *AgentPolicyService) Set(ctx context.Context, portalUserID string, polic if s == nil || s.pool == nil { return errors.New("安全策略服务不可用") } - if policy.RateLimitMultiplier < 1 || policy.RateLimitMultiplier > 100 { - return errors.New("限流倍数必须在 1-100 之间") + if policy.RateLimitMultiplier < 1 || policy.RateLimitMultiplier > 10 { + return errors.New("限流倍数必须在 1-10 之间") } _, err := s.pool.Exec(ctx, `INSERT INTO gateway.portal_agent_policies(portal_user_id,auto_approve_tools,rate_limit_multiplier) VALUES($1,$2,$3) ON CONFLICT(portal_user_id) DO UPDATE SET auto_approve_tools=$2,rate_limit_multiplier=$3,updated_at=clock_timestamp()`, diff --git a/internal/workbench/tools.go b/internal/workbench/tools.go index 4b53b26..12371dc 100644 --- a/internal/workbench/tools.go +++ b/internal/workbench/tools.go @@ -11,6 +11,7 @@ import ( "net/http" "net/url" "strings" + "syscall" "time" "aigateway.local/core/internal/platform/cryptox" @@ -184,6 +185,14 @@ func (s *ToolService) Delete(ctx context.Context, id, actorID string) error { if used { return ErrConflict } + // 有调用历史的工具禁止删除:tool_runs 级联删除会永久丢失报表/审计数据。 + var hasRuns bool + if err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.tool_runs WHERE tool_id=$1)`, id).Scan(&hasRuns); err != nil { + return err + } + if hasRuns { + return errors.New("工具存在调用历史,不能删除;请停用以保留报表数据") + } tag, err := tx.Exec(ctx, `DELETE FROM gateway.tool_definitions WHERE id=$1`, id) if err != nil { return err @@ -234,7 +243,12 @@ func (s *ToolService) enforceGovernance(ctx context.Context, tool Tool, apiKeyID if err := s.assets.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.tool_approval_requests WHERE tool_id=$1 AND status='approved')`, tool.ID).Scan(&approved); err != nil { return false, err } - if !approved { + // rejected 后 24 小时冷却:避免每次调用都重新发起申请、通知轰炸管理员。 + var recentlyRejected bool + if err := s.assets.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.tool_approval_requests WHERE tool_id=$1 AND status='rejected' AND decided_at>clock_timestamp()-interval '24 hours')`, tool.ID).Scan(&recentlyRejected); err != nil { + return false, err + } + if !approved && !recentlyRejected { // 自动发起待审申请(唯一部分索引防重复),通知管理员。 requestID, err := newUUID() if err != nil { @@ -283,6 +297,9 @@ func (s *ToolService) enforceGovernance(ctx context.Context, tool Tool, apiKeyID return false, err } if count > int64(limit) { + // 被拒调用补偿递减:该次计数不消耗窗口额度,避免故障重试风暴 + // 打满窗口波及合法调用者。 + _, _ = s.assets.pool.Exec(ctx, `UPDATE gateway.tool_rate_usage SET call_count=greatest(call_count-1,0) WHERE tool_id=$1 AND portal_user_id=$2 AND window_start=date_trunc('minute',clock_timestamp())`, tool.ID, userKey) return false, ErrToolRateLimited } } @@ -510,24 +527,21 @@ func safeToolDial(allowPrivate bool) func(context.Context, string, string) (net. if allowPrivate { return dialer.DialContext } - return func(ctx context.Context, network, address string) (net.Conn, error) { - host, port, err := net.SplitHostPort(address) + // 在系统 connect 阶段用 Dialer.Control 校验最终目标 IP:内核完成 + // 解析后、TCP 握手前回调,校验与连接之间不存在 DNS rebinding 窗口。 + dialer.Control = func(_, address string, _ syscall.RawConn) error { + host, _, err := net.SplitHostPort(address) if err != nil { - return nil, err + return err } - addresses, err := net.DefaultResolver.LookupIPAddr(ctx, host) - if err != nil { - return nil, err + ip := net.ParseIP(host) + if ip == nil { + return errors.New("工具目标不是 IP 地址") } - if len(addresses) == 0 { - return nil, errors.New("工具主机没有解析结果") + if ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsMulticast() || ip.IsUnspecified() { + return fmt.Errorf("工具目标为受限地址 %s", ip) } - for _, candidate := range addresses { - ip := candidate.IP - if ip == nil || ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsMulticast() || ip.IsUnspecified() { - return nil, fmt.Errorf("工具主机解析到受限地址 %s", ip) - } - } - return dialer.DialContext(ctx, network, net.JoinHostPort(addresses[0].IP.String(), port)) + return nil } + return dialer.DialContext } diff --git a/web/apps/admin/src/views/gateway/channels/index.vue b/web/apps/admin/src/views/gateway/channels/index.vue index acebcd2..a37ee12 100644 --- a/web/apps/admin/src/views/gateway/channels/index.vue +++ b/web/apps/admin/src/views/gateway/channels/index.vue @@ -75,7 +75,10 @@ - + + + +