superidou
9501751792
三轮审查修复(60+ 项),相对远端 main(b536672)的关键变更:
- 安全: 数据面 SSRF 拨号防护(防 DNS rebinding)/上游凭据剥离/登录防枚举
与锁定态统一/可信代理(X-Forwarded-For)限流加固/会话版本失效机制/
撤销即时传播/弱密钥拒绝启动/脱敏字节级重写(保签名契约)
- 业务逻辑: 裸 body 上传 panic/bootstrap 审计管线卡死/定价通配符优先级/
全局工具可见性/调度器停机补跑/TOTP 挑战令牌消费顺序/熔断探针语义/
>4MB 响应 token 计量/管理员重置密码作废会话 等
- 前端: 新 logo(语枢 AI 网关主题)/Provider 凭据异常警示/删除入口/
后端错误消息透传/localStorage 敏感数据收敛
- 部署: CREDENTIAL_MASTER_KEY 持久化与弱值拒绝/Provider DELETE 接口/
nginx 安全头/worker 内存限制
- 新增迁移 000029(key_hash 索引)/000030(usage_daily 币种维度)
386 lines
13 KiB
Go
386 lines
13 KiB
Go
package workbench
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
func validateApplication(app *Application) error {
|
|
app.Code = strings.ToLower(strings.TrimSpace(app.Code))
|
|
app.Name = strings.TrimSpace(app.Name)
|
|
app.Description = strings.TrimSpace(app.Description)
|
|
app.Status = strings.TrimSpace(app.Status)
|
|
if !codeRE.MatchString(app.Code) || len(app.Code) < 3 || app.Name == "" || len(app.Name) > 128 || len(app.Description) > 4000 {
|
|
return errors.New("应用编码、名称或描述格式无效")
|
|
}
|
|
switch app.Status {
|
|
case "draft", "active", "suspended", "retired":
|
|
case "":
|
|
app.Status = "draft"
|
|
default:
|
|
return errors.New("应用状态无效")
|
|
}
|
|
var err error
|
|
app.DepartmentIDs, err = normalizeStrings(app.DepartmentIDs, 100)
|
|
return err
|
|
}
|
|
|
|
func normalizeApplicationConfig(config *ApplicationConfig, requireModel bool) error {
|
|
config.Model = strings.TrimSpace(config.Model)
|
|
config.PromptTemplateID = strings.TrimSpace(config.PromptTemplateID)
|
|
if requireModel && config.Model == "" {
|
|
return errors.New("发布配置必须指定模型")
|
|
}
|
|
if len(config.Model) > 255 {
|
|
return errors.New("模型名称过长")
|
|
}
|
|
var err error
|
|
if config.KnowledgeBaseIDs, err = normalizeStrings(config.KnowledgeBaseIDs, 20); err != nil {
|
|
return err
|
|
}
|
|
if config.ToolIDs, err = normalizeStrings(config.ToolIDs, 20); err != nil {
|
|
return err
|
|
}
|
|
if config.RetrievalTopK == 0 {
|
|
config.RetrievalTopK = 4
|
|
}
|
|
if config.RetrievalTopK < 1 || config.RetrievalTopK > 20 {
|
|
return errors.New("retrieval_top_k 应在 1-20 之间")
|
|
}
|
|
if config.Temperature < 0 || config.Temperature > 2 {
|
|
return errors.New("temperature 应在 0-2 之间")
|
|
}
|
|
// 缺省 max_tool_rounds(0)时按 5 轮处理,与数字员工一致;否则运行时
|
|
// round >= 0 在第一次工具调用前就判定"已达上限",应用永远无法完成工具调用。
|
|
if config.MaxToolRounds == 0 {
|
|
config.MaxToolRounds = 5
|
|
}
|
|
if config.MaxToolRounds < 1 || config.MaxToolRounds > 8 {
|
|
return errors.New("max_tool_rounds 应在 1-8 之间")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) validateApplicationRefs(ctx context.Context, config ApplicationConfig) error {
|
|
if config.PromptTemplateID != "" {
|
|
var ok bool
|
|
if err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.prompt_templates WHERE id=$1 AND enabled AND current_version IS NOT NULL)`, config.PromptTemplateID).Scan(&ok); err != nil {
|
|
return err
|
|
}
|
|
if !ok {
|
|
return errors.New("绑定的 Prompt 不存在、未启用或没有生效版本")
|
|
}
|
|
}
|
|
for _, id := range config.KnowledgeBaseIDs {
|
|
var ok bool
|
|
if err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.knowledge_bases WHERE id=$1 AND enabled)`, id).Scan(&ok); err != nil {
|
|
return err
|
|
}
|
|
if !ok {
|
|
return fmt.Errorf("知识库 %s 不存在或未启用", id)
|
|
}
|
|
}
|
|
for _, id := range config.ToolIDs {
|
|
var ok bool
|
|
if err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.tool_definitions WHERE id=$1 AND enabled)`, id).Scan(&ok); err != nil {
|
|
return err
|
|
}
|
|
if !ok {
|
|
return fmt.Errorf("工具 %s 不存在或未启用", id)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
const applicationSelect = `SELECT a.id::text,a.code,a.name,a.description,a.department_ids::text[],a.status,a.draft_config,a.published_version,p.config,a.revision,a.created_at,a.updated_at FROM gateway.applications a LEFT JOIN gateway.application_versions p ON p.application_id=a.id AND p.version=a.published_version`
|
|
|
|
func scanApplication(row pgx.Row) (Application, error) {
|
|
var app Application
|
|
var draft []byte
|
|
var published []byte
|
|
err := row.Scan(&app.ID, &app.Code, &app.Name, &app.Description, &app.DepartmentIDs, &app.Status, &draft, &app.PublishedVersion, &published, &app.Revision, &app.CreatedAt, &app.UpdatedAt)
|
|
if err != nil {
|
|
return app, mapNotFound(err)
|
|
}
|
|
_ = json.Unmarshal(draft, &app.DraftConfig)
|
|
if len(published) > 0 {
|
|
var config ApplicationConfig
|
|
if json.Unmarshal(published, &config) == nil {
|
|
app.PublishedConfig = &config
|
|
}
|
|
}
|
|
return app, nil
|
|
}
|
|
func (s *Service) ListApplications(ctx context.Context) ([]Application, error) {
|
|
rows, err := s.pool.Query(ctx, applicationSelect+` ORDER BY a.updated_at DESC`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []Application{}
|
|
for rows.Next() {
|
|
app, err := scanApplication(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, app)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
func (s *Service) GetApplication(ctx context.Context, id string) (Application, error) {
|
|
return scanApplication(s.pool.QueryRow(ctx, applicationSelect+` WHERE a.id=$1`, id))
|
|
}
|
|
func (s *Service) GetPublishedApplicationByCode(ctx context.Context, code string) (Application, error) {
|
|
return scanApplication(s.pool.QueryRow(ctx, applicationSelect+` WHERE a.code=$1 AND a.status='active' AND a.published_version IS NOT NULL`, code))
|
|
}
|
|
|
|
func (s *Service) SaveApplication(ctx context.Context, app Application, actorID string, create bool) (Application, error) {
|
|
if err := validateApplication(&app); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if err := normalizeApplicationConfig(&app.DraftConfig, false); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if err := s.validateApplicationRefs(ctx, app.DraftConfig); err != nil {
|
|
return Application{}, err
|
|
}
|
|
raw, _ := json.Marshal(app.DraftConfig)
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return Application{}, err
|
|
}
|
|
defer rollback(ctx, tx)
|
|
if create {
|
|
app.ID, err = newUUID()
|
|
if err != nil {
|
|
return Application{}, err
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO gateway.applications(id,code,name,description,department_ids,status,draft_config,created_by) VALUES($1,$2,$3,$4,$5,$6,$7,$8)`, app.ID, app.Code, app.Name, app.Description, app.DepartmentIDs, app.Status, raw, actorID)
|
|
} else {
|
|
tag, updateErr := tx.Exec(ctx, `UPDATE gateway.applications SET code=$2,name=$3,description=$4,department_ids=$5,status=$6,draft_config=$7,revision=revision+1,updated_at=clock_timestamp() WHERE id=$1`, app.ID, app.Code, app.Name, app.Description, app.DepartmentIDs, app.Status, raw)
|
|
err = updateErr
|
|
if err == nil && tag.RowsAffected() == 0 {
|
|
return Application{}, ErrNotFound
|
|
}
|
|
}
|
|
if err != nil {
|
|
return Application{}, err
|
|
}
|
|
event := "application.updated"
|
|
if create {
|
|
event = "application.created"
|
|
}
|
|
if err = emit(ctx, tx, event, "application", app.ID, actorID, nil); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if err = tx.Commit(ctx); err != nil {
|
|
return Application{}, err
|
|
}
|
|
return s.GetApplication(ctx, app.ID)
|
|
}
|
|
|
|
func (s *Service) PublishApplication(ctx context.Context, id, changeNote, actorID string) (Application, error) {
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return Application{}, err
|
|
}
|
|
defer rollback(ctx, tx)
|
|
var raw []byte
|
|
if err = tx.QueryRow(ctx, `SELECT draft_config FROM gateway.applications WHERE id=$1 FOR UPDATE`, id).Scan(&raw); errors.Is(err, pgx.ErrNoRows) {
|
|
return Application{}, ErrNotFound
|
|
} else if err != nil {
|
|
return Application{}, err
|
|
}
|
|
var config ApplicationConfig
|
|
if err = json.Unmarshal(raw, &config); err != nil {
|
|
return Application{}, errors.New("应用草稿配置无效")
|
|
}
|
|
if err = normalizeApplicationConfig(&config, true); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if err = s.validateApplicationRefs(ctx, config); err != nil {
|
|
return Application{}, err
|
|
}
|
|
raw, _ = json.Marshal(config)
|
|
var version int
|
|
if err = tx.QueryRow(ctx, `SELECT coalesce(max(version),0)+1 FROM gateway.application_versions WHERE application_id=$1`, id).Scan(&version); err != nil {
|
|
return Application{}, err
|
|
}
|
|
versionID, idErr := newUUID()
|
|
if idErr != nil {
|
|
return Application{}, idErr
|
|
}
|
|
if _, err = tx.Exec(ctx, `INSERT INTO gateway.application_versions(id,application_id,version,config,change_note,published_by) VALUES($1,$2,$3,$4,$5,$6)`, versionID, id, version, raw, strings.TrimSpace(changeNote), actorID); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if _, err = tx.Exec(ctx, `UPDATE gateway.applications SET published_version=$2,status='active',revision=revision+1,updated_at=clock_timestamp() WHERE id=$1`, id, version); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if err = emit(ctx, tx, "application.published", "application", id, actorID, map[string]any{"version": version}); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if err = tx.Commit(ctx); err != nil {
|
|
return Application{}, err
|
|
}
|
|
return s.GetApplication(ctx, id)
|
|
}
|
|
|
|
func (s *Service) DeleteApplication(ctx context.Context, id, actorID string) error {
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rollback(ctx, tx)
|
|
tag, err := tx.Exec(ctx, `DELETE FROM gateway.applications WHERE id=$1`, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNotFound
|
|
}
|
|
if err = emit(ctx, tx, "application.deleted", "application", id, actorID, nil); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (s *Service) ListApplicationVersions(ctx context.Context, id string) ([]map[string]any, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT id::text,version,config,change_note,created_at FROM gateway.application_versions WHERE application_id=$1 ORDER BY version DESC`, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []map[string]any{}
|
|
for rows.Next() {
|
|
var versionID, note string
|
|
var version int
|
|
var raw []byte
|
|
var created time.Time
|
|
if err = rows.Scan(&versionID, &version, &raw, ¬e, &created); err != nil {
|
|
return nil, err
|
|
}
|
|
var config ApplicationConfig
|
|
_ = json.Unmarshal(raw, &config)
|
|
items = append(items, map[string]any{"id": versionID, "version": version, "config": config, "change_note": note, "created_at": created})
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) ListApplicationRuns(ctx context.Context, id string, limit int) ([]ApplicationRun, error) {
|
|
if limit < 1 {
|
|
limit = 50
|
|
}
|
|
if limit > 200 {
|
|
limit = 200
|
|
}
|
|
rows, err := s.pool.Query(ctx, `SELECT id::text,application_id::text,request_id,status,error,version,latency_ms,retrieval_count,tool_count,created_at FROM gateway.application_runs WHERE application_id=$1 ORDER BY created_at DESC LIMIT $2`, id, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ApplicationRun{}
|
|
for rows.Next() {
|
|
var run ApplicationRun
|
|
if err = rows.Scan(&run.ID, &run.ApplicationID, &run.RequestID, &run.Status, &run.Error, &run.Version, &run.LatencyMS, &run.RetrievalCount, &run.ToolCount, &run.CreatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, run)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) RollbackApplication(ctx context.Context, id string, version int, changeNote, actorID string) (Application, error) {
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return Application{}, err
|
|
}
|
|
defer rollback(ctx, tx)
|
|
var raw []byte
|
|
if err = tx.QueryRow(ctx, `SELECT config FROM gateway.application_versions WHERE application_id=$1 AND version=$2`, id, version).Scan(&raw); errors.Is(err, pgx.ErrNoRows) {
|
|
return Application{}, ErrNotFound
|
|
} else if err != nil {
|
|
return Application{}, err
|
|
}
|
|
var config ApplicationConfig
|
|
if err = json.Unmarshal(raw, &config); err != nil {
|
|
return Application{}, errors.New("目标版本配置无效")
|
|
}
|
|
if err = normalizeApplicationConfig(&config, true); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if err = s.validateApplicationRefs(ctx, config); err != nil {
|
|
return Application{}, err
|
|
}
|
|
var next int
|
|
if err = tx.QueryRow(ctx, `SELECT coalesce(max(version),0)+1 FROM gateway.application_versions WHERE application_id=$1`, id).Scan(&next); err != nil {
|
|
return Application{}, err
|
|
}
|
|
versionID, err := newUUID()
|
|
if err != nil {
|
|
return Application{}, err
|
|
}
|
|
note := strings.TrimSpace(changeNote)
|
|
if note == "" {
|
|
note = fmt.Sprintf("回滚到 v%d", version)
|
|
}
|
|
if _, err = tx.Exec(ctx, `INSERT INTO gateway.application_versions(id,application_id,version,config,change_note,published_by) VALUES($1,$2,$3,$4,$5,$6)`, versionID, id, next, raw, note, actorID); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if _, err = tx.Exec(ctx, `UPDATE gateway.applications SET draft_config=$2,published_version=$3,status='active',revision=revision+1,updated_at=clock_timestamp() WHERE id=$1`, id, raw, next); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if err = emit(ctx, tx, "application.rolled_back", "application", id, actorID, map[string]any{"source_version": version, "published_version": next}); err != nil {
|
|
return Application{}, err
|
|
}
|
|
if err = tx.Commit(ctx); err != nil {
|
|
return Application{}, err
|
|
}
|
|
return s.GetApplication(ctx, id)
|
|
}
|
|
|
|
func (s *Service) ApplicationDependencies(ctx context.Context, assetType, assetID string) ([]map[string]any, error) {
|
|
apps, err := s.ListApplications(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items := []map[string]any{}
|
|
for _, app := range apps {
|
|
stages := []struct {
|
|
name string
|
|
config *ApplicationConfig
|
|
}{{"draft", &app.DraftConfig}, {"published", app.PublishedConfig}}
|
|
for _, stage := range stages {
|
|
if stage.config == nil {
|
|
continue
|
|
}
|
|
used := assetType == "prompt" && stage.config.PromptTemplateID == assetID
|
|
if assetType == "knowledge" {
|
|
for _, id := range stage.config.KnowledgeBaseIDs {
|
|
if id == assetID {
|
|
used = true
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if assetType == "tool" {
|
|
for _, id := range stage.config.ToolIDs {
|
|
if id == assetID {
|
|
used = true
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if used {
|
|
items = append(items, map[string]any{"application_id": app.ID, "code": app.Code, "name": app.Name, "stage": stage.name, "version": app.PublishedVersion})
|
|
}
|
|
}
|
|
}
|
|
return items, nil
|
|
}
|