From 20d226854675f6e9bd4ba51a457e1566f013efe8 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 11 Aug 2026 15:21:05 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20close=20P10=20audit=20items=20=E2=80=94?= =?UTF-8?q?=20P10-1=20sources=20API=20admin=20guard,=20P10-2=20scope=20pre?= =?UTF-8?q?fix=20strip,=20P10-3=20runtime=20source=20timeout=20w/=20stream?= =?UTF-8?q?-safe=20clients?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - api.go: handleSourcesAPI now requires admin role (GET leaks upstream api_keys, POST/DELETE mutate routing) - chat.go: hasScopeModel made a Gateway method that strips source-model/:// prefix strictly via Registry.EffectiveModel (only when the prefix names a real source serving the bare model) so dash-bearing ids like deepseek-v4-flash-free are never corrupted; +TestHasScopeModelWithSourcePrefix - config.go: DefaultSourceTimeout/QueueTimeout/Concurrency constants shared by YAML ApplyDefaults and runtime sources - core.go: mergedSources applies the same defaults to runtime sources (JSON never persisted timeout fields); a dead upstream can no longer hold a concurrency slot forever - provider.go: split non-streaming client{Timeout} vs stream client{} sharing a Transport with ResponseHeaderTimeout, so long SSE bodies are not cut by client.Timeout; ChatStream uses doRawStream - plan.md: mark P4-4/5/6 done, record P4-7/8 (tier-order, audit export, UI key view, zen upstream diagnosis) - online verified: user key -> /api/sources 403 (GET+POST), admin 200, AUTO stream/non-stream healthy --- .gitignore | 2 +- internal/config/config.go | 19 +++++++++----- internal/core/core.go | 15 ++++++++++- internal/gateway/api.go | 7 +++++- internal/gateway/chat.go | 15 ++++++++--- internal/gateway/gateway_test.go | 35 +++++++++++++++++++++++++- internal/provider/provider.go | 43 +++++++++++++++++++++++++++++--- plan.md | 8 +++--- 8 files changed, 125 insertions(+), 19 deletions(-) diff --git a/.gitignore b/.gitignore index 28f49e0..f2c99db 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,4 @@ /llmsproxy /config.yaml /adapters/ -*.log \ No newline at end of file +*.logmaster.key diff --git a/internal/config/config.go b/internal/config/config.go index b9324f5..e62a4e7 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -19,13 +19,20 @@ type Config struct { DefaultModel string `yaml:"default_model"` // e.g. "AUTO" or a model id AdapterDir string `yaml:"adapter_dir"` RuntimeFile string `yaml:"runtime_file"` - MaxConcurrent int `yaml:"max_concurrent"` // global inflight cap, 0 = unlimited - TLSCertFile string `yaml:"tls_cert_file,omitempty"` // PEM cert; when set together with tls_key_file, serve HTTPS - TLSKeyFile string `yaml:"tls_key_file,omitempty"` // PEM private key + MaxConcurrent int `yaml:"max_concurrent"` // global inflight cap, 0 = unlimited + TLSCertFile string `yaml:"tls_cert_file,omitempty"` // PEM cert; when set together with tls_key_file, serve HTTPS + TLSKeyFile string `yaml:"tls_key_file,omitempty"` // PEM private key PublicBaseURL string `yaml:"public_base_url,omitempty"` // external base for generated config snippets; default inferred from request Sources []Source `yaml:"sources"` } +// Defaults applied to any source (YAML or runtime) that leaves a field unset. +const ( + DefaultSourceTimeout = 120 * time.Second + DefaultSourceQueueTimeout = 60 * time.Second + DefaultSourceConcurrency = 8 +) + // Model is a single exposed model id bound to a source, with priority used by // AUTO auto selection (higher number = preferred). type Model struct { @@ -149,13 +156,13 @@ func (c *Config) ApplyDefaults() error { s.Adapter = "openai" } if s.Timeout == 0 { - s.Timeout = 120 * time.Second + s.Timeout = DefaultSourceTimeout } if s.QueueTimeout == 0 { - s.QueueTimeout = 60 * time.Second + s.QueueTimeout = DefaultSourceQueueTimeout } if s.MaxConcurrent == 0 { - s.MaxConcurrent = 8 + s.MaxConcurrent = DefaultSourceConcurrency } if seen[s.Name] { return fmt.Errorf("config: duplicate source name %q", s.Name) diff --git a/internal/core/core.go b/internal/core/core.go index 114ed12..c4768c5 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -298,7 +298,20 @@ func (c *Core) mergedSources() []config.Source { for _, n := range order { if !seen[n] { seen[n] = true - out = append(out, c.resolveSourceKey(byName[n])) + s := c.resolveSourceKey(byName[n]) + // Runtime sources (web UI edits) are persisted without timeout + // fields; apply the same defaults the YAML path gets so a dead + // upstream cannot hold a concurrency slot forever (P10-3). + if s.Timeout == 0 { + s.Timeout = config.DefaultSourceTimeout + } + if s.QueueTimeout == 0 { + s.QueueTimeout = config.DefaultSourceQueueTimeout + } + if s.MaxConcurrent == 0 { + s.MaxConcurrent = config.DefaultSourceConcurrency + } + out = append(out, s) } } return out diff --git a/internal/gateway/api.go b/internal/gateway/api.go index 7cb0da5..76c1e1f 100644 --- a/internal/gateway/api.go +++ b/internal/gateway/api.go @@ -75,6 +75,10 @@ type sourcePayload struct { } func (g *Gateway) handleSourcesAPI(w http.ResponseWriter, r *http.Request) { + if reqRole(r.Context()) != "admin" { + writeError(w, http.StatusForbidden, "forbidden", "admin role required") + return + } path := strings.TrimPrefix(r.URL.Path, "/api/sources") path = strings.Trim(path, "/") @@ -121,6 +125,7 @@ func (g *Gateway) handleSourcesAPI(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "") } } + // handleStatsAPI returns per-key / per-model / per-source usage aggregates and // the recent request audit trail. func (g *Gateway) handleStatsAPI(w http.ResponseWriter, r *http.Request) { @@ -173,7 +178,7 @@ func (g *Gateway) handleStatsAPI(w http.ResponseWriter, r *http.Request) { cw.Flush() return } -if r.URL.Query().Get("export") == "keys-csv" { + if r.URL.Query().Get("export") == "keys-csv" { w.Header().Set("Content-Type", "text/csv; charset=utf-8") w.Header().Set("Content-Disposition", "attachment; filename=llmsproxy-keys.csv") cw := csv.NewWriter(w) diff --git a/internal/gateway/chat.go b/internal/gateway/chat.go index c1a4e16..8c43b5c 100644 --- a/internal/gateway/chat.go +++ b/internal/gateway/chat.go @@ -216,7 +216,16 @@ func (g *Gateway) scopeTokens(ctx context.Context, sc config.ModelScope) int64 { return g.stats.WindowTokens(sc.Model, sc.Source, win) } -func hasScopeModel(list []config.ModelScope, s string) bool { +// hasScopeModel reports whether a model (possibly with a "source-model" / +// "source:model" / "source/model" pinning prefix) is allowed by a key's model +// scope. The prefix is stripped strictly: only when the prefix names a real +// source that actually serves the bare model (via Registry.EffectiveModel), so +// model ids that themselves contain separators (e.g. "deepseek-v4-flash-free") +// are never corrupted (P10-2). +func (g *Gateway) hasScopeModel(list []config.ModelScope, s string) bool { + if r := g.core.Registry(); r != nil { + s = r.EffectiveModel(s) + } for _, x := range list { if x.Model == s || (x.Model != "" && strings.EqualFold(x.Model, "AUTO")) { return true @@ -325,7 +334,7 @@ func (g *Gateway) handleChat(w http.ResponseWriter, r *http.Request) { return } if !isAuto(model) { - if allow := g.allowedModels(r.Context()); allow != nil && !hasScopeModel(allow, model) { + if allow := g.allowedModels(r.Context()); allow != nil && !g.hasScopeModel(allow, model) { writeError(w, http.StatusForbidden, "model_not_allowed", fmt.Sprintf("model %q is not allowed for this key", model)) return } @@ -811,7 +820,7 @@ func (g *Gateway) handleImage(w http.ResponseWriter, r *http.Request) { model = g.core.DefaultModel() } if !isAuto(model) { - if allow := g.allowedModels(r.Context()); allow != nil && !hasScopeModel(allow, model) { + if allow := g.allowedModels(r.Context()); allow != nil && !g.hasScopeModel(allow, model) { writeError(w, http.StatusForbidden, "model_not_allowed", fmt.Sprintf("model %q is not allowed for this key", model)) return } diff --git a/internal/gateway/gateway_test.go b/internal/gateway/gateway_test.go index 02b9a78..a5f4674 100644 --- a/internal/gateway/gateway_test.go +++ b/internal/gateway/gateway_test.go @@ -195,7 +195,7 @@ func TestAutoStatesReportChainHealth(t *testing.T) { t.Fatalf("get auto status=%d body=%s", rr.Code, rr.Body.String()) } var body struct { - Rules []config.ModelScope `json:"rules"` + Rules []config.ModelScope `json:"rules"` States []core.AutoSlotState `json:"states"` } if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { @@ -634,3 +634,36 @@ func TestAPIChatInternal(t *testing.T) { t.Fatalf("api chat body=%s", rr.Body.String()) } } + +func TestHasScopeModelWithSourcePrefix(t *testing.T) { + // zen serves deepseek-v4-flash-free, so the pinning prefix strips; deepseek + // serves deepseek-v4-pro but NOT v4-flash-free, so a model id like + // "deepseek-v4-flash-free" (source deepseek + model id with dashes) must NOT + // be corrupted by prefix stripping. + g := newTestGateway(t, + config.Source{Name: "deepseek", BaseURL: "http://d", Adapter: "openai", + Models: []config.Model{{ID: "deepseek-v4-pro"}}}, + config.Source{Name: "zen", BaseURL: "http://z", Adapter: "openai", + Models: []config.Model{{ID: "deepseek-v4-flash-free"}}}, + ) + scope := []config.ModelScope{{Model: "deepseek-v4-flash-free"}, {Model: "gpt-5.6-sol"}} + for _, prefixed := range []string{"zen:deepseek-v4-flash-free", "zen/deepseek-v4-flash-free"} { + if !g.hasScopeModel(scope, prefixed) { + t.Errorf("hasScopeModel(scope, %q)=false, want true (prefix must be stripped)", prefixed) + } + } + for _, bare := range []string{"deepseek-v4-flash-free", "gpt-5.6-sol"} { + if !g.hasScopeModel(scope, bare) { + t.Errorf("hasScopeModel(scope, %q)=false, want true", bare) + } + } + if g.hasScopeModel(scope, "deepseek-v4-pro") { + t.Error("hasScopeModel returned true for model outside scope") + } + if g.hasScopeModel(scope, "deepseek-v4-flash-free-extra") { + t.Error("hasScopeModel returned true for unrelated model") + } + if !g.hasScopeModel([]config.ModelScope{{Model: "AUTO"}}, "zen:anything") { + t.Error("AUTO scope should allow any prefixed model") + } +} diff --git a/internal/provider/provider.go b/internal/provider/provider.go index 2a5467f..239f932 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -10,6 +10,7 @@ import ( "encoding/json" "fmt" "io" + "net" "net/http" "strings" "sync" @@ -129,7 +130,11 @@ type Provider struct { cfg config.Source vm *lua.VM adapter string - client *http.Client + // client bounds a whole non-streaming request (dial+read body). stream + // is used for SSE: no overall timeout (a long stream must not be cut), + // only the transport's ResponseHeaderTimeout bounds time-to-first-byte. + client *http.Client + stream *http.Client mu sync.Mutex sem chan struct{} @@ -142,11 +147,31 @@ type Provider struct { } func New(cfg config.Source, vm *lua.VM) *Provider { + if cfg.Timeout <= 0 { + cfg.Timeout = config.DefaultSourceTimeout + } + if cfg.MaxConcurrent <= 0 { + cfg.MaxConcurrent = config.DefaultSourceConcurrency + } + // Shared transport: ResponseHeaderTimeout bounds how long we wait for the + // first response byte (applies to both paths); the stream client has no + // client-level Timeout so the SSE body can run past the header timeout. + tr := &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: (&net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second}).DialContext, + ForceAttemptHTTP2: true, + MaxIdleConns: 100, + IdleConnTimeout: 90 * time.Second, + TLSHandshakeTimeout: 10 * time.Second, + ExpectContinueTimeout: 1 * time.Second, + ResponseHeaderTimeout: cfg.Timeout, + } p := &Provider{ cfg: cfg, vm: vm, adapter: cfg.Adapter, - client: &http.Client{Timeout: cfg.Timeout}, + client: &http.Client{Timeout: cfg.Timeout, Transport: tr}, + stream: &http.Client{Transport: tr}, sem: make(chan struct{}, cfg.MaxConcurrent), states: map[string]*ModelState{}, } @@ -632,7 +657,7 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch } rc := make(chan respOrErr, 1) go func() { - resp, err := p.doRaw(ctx, p.URL(), body, hdrs) + resp, err := p.doRawStream(ctx, p.URL(), body, hdrs) rc <- respOrErr{resp, err} }() @@ -770,6 +795,18 @@ func (p *Provider) doRaw(ctx context.Context, url, body string, hdr http.Header) return p.client.Do(httpReq) } +// doRawStream is the streaming variant of doRaw: it uses the no-overall-timeout +// stream client so a long SSE body is not cut by client.Timeout. The transport +// still enforces ResponseHeaderTimeout on time-to-first-byte. +func (p *Provider) doRawStream(ctx context.Context, url, body string, hdr http.Header) (*http.Response, error) { + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader([]byte(body))) + if err != nil { + return nil, err + } + httpReq.Header = hdr + return p.stream.Do(httpReq) +} + func marshalTransform(vm *lua.VM, adapter, fn string, v interface{}) (string, error) { b, err := json.Marshal(v) if err != nil { diff --git a/plan.md b/plan.md index 83d5e93..a393ee8 100644 --- a/plan.md +++ b/plan.md @@ -240,6 +240,8 @@ p==nil → 400/404；!TryAcquire → 429 busy（快速失败） - [x] **P4-1（2026-08-11）WebUI 聊天"每次回复都失败"（用户实测）**：根因——`sendChat()` 调 `addMsg('assistant', '')` 未传 reason 参数 → `thinkEl`/`hintEl` 为 null → 首个携带 `reasoning_content` 的流式 chunk 到达时 index.html 抛 `Cannot set properties of null`，前端表现为请求失败。修复：`addMsg('assistant', '', true)` + 初始隐藏 think 框；提交 `7647b7d`，已推送并部署生产（服务 active、`/api/status` 正常），`go test ./internal/gateway/...` 通过。 - [x] **P4-2（2026-08-11）41s AUTO 延迟归因**：审计 `09:54:42 stream lat=41810 model=deepseek-v4-flash-free src=zen ok=200 pt=17 ct=109`——链上调度仅 ~0.6s（tier1 qijiar 失败 lat=657ms 后顺延），其余 ~41s 为 zen 上游生成耗时（~380ms/token），**非网关调度问题**。 - [x] **P4-3（2026-08-11）全量代码审查**：provider.go（812 行）/stats.go（459 行）/scheduler.go（398 行）/registry.go（258 行）/api.go/chat.go/server.go/config 全部通读完毕，确认 P10-1~P10-5（见上）。 -- [ ] **P4-4**：`handleSourcesAPI` 补 admin 校验（P10-1，一行仿 api.go:21）— 待用户确认后实施。 -- [ ] **P4-5**：`hasScopeModel` 剥前缀比对（P10-2）— 待确认。 -- [ ] **P4-6**：runtime 源默认超时策略（P10-3，区分流式/非流式）— 待确认。 \ No newline at end of file +- [x] **P4-4（2026-08-11）`handleSourcesAPI` 补 admin 校验（P10-1）**：`api.go` 方法开头加 `reqRole(r.Context()) != "admin"` 检查（同 handleAdaptersAPI）；线上实测 user 级 key GET/POST `/api/sources` 均 403，admin 200，防越权读取明文 api_key 与篡改源。 +- [x] **P4-5（2026-08-11）`hasScopeModel` 剥前缀比对（P10-2）**：改为 Gateway 方法，先用 `Registry.EffectiveModel` 严格剥 `source-`/`source:`/`source/` 前缀（仅当前缀是真实源名且该源确实服务裸模型时才剥，避免误伤 `deepseek-v4-flash-free` 这类自带 `-` 的模型 ID）；单测 `TestHasScopeModelWithSourcePrefix` 覆盖前缀剥离与勿误伤。 +- [x] **P4-6（2026-08-11）runtime 源默认超时策略（P10-3，区分流式/非流式）**：新增 `config.DefaultSourceTimeout=120s/DefaultSourceQueueTimeout=60s/DefaultSourceConcurrency=8`；`Core.mergedSources` 对 runtime 源兜底（JSON 源不序列化 timeout）；`Provider.New` 拆两个 client——非流式 `client{Timeout}` 整请求限时、流式 `stream` 无总超时（共享 Transport 仅 `ResponseHeaderTimeout` 限首字节），长 SSE 不被掐；`ChatStream` 改用 `doRawStream`。单测 + 线上流/非流式验证通过。 +- [x] **P4-7（2026-08-11）AUTO 链 tier 序 + 审计导出修复 + UI 密钥视图（今日会话）**：① `BuildChain` 排序由降序改升序（tier1=最高优先级先试，修 P10-5 中的序颠倒），单测同步；② 统计/CSV 全量（limit 20000、`Stats.AuditRecords` 读磁盘含 `*.old`、key_names 掩码键、非 admin 过滤掩码）；③ UI 卡片滚动区修正 + 密钥视图 toggle（再点同一密钥回全局）+ `rec-exit` 退出入口；④ WebUI 测试"所有档位失败"观察：AUTO 链降级正常（opus 必死→deepseek 撞 zen 间歇故障→frank key 失效→nemotron 404 全灭 503）。 +- [x] **P4-8（2026-08-11）调度延滞排查结论**：homeagent AUTO 请求"看起来停在 opus"实为调度正常降级——完整 ChainErr 列出 tier1→2→3→4 全部尝试，审计只记 `ce.Tiers[0]`（首档）故 UI 显 opus；zen 源 `opencode.ai/zen/v1` 直连即 403 `[server_error] Upstream response was not valid JSON`（间歇）/ nemotron 404，确认为上游源自身问题，非网关透传或 homeagent 适配器。 \ No newline at end of file