fix: close P10 audit items — P10-1 sources API admin guard, P10-2 scope prefix strip, P10-3 runtime source timeout w/ stream-safe clients

- 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
This commit is contained in:
root
2026-08-11 15:21:05 +08:00
parent 854b3e2e37
commit 20d2268546
8 changed files with 125 additions and 19 deletions

View File

@ -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)

View File

@ -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

View File

@ -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)

View File

@ -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
}

View File

@ -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")
}
}

View File

@ -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 {