diff --git a/admin.go b/admin.go index b846c59..382a043 100644 --- a/admin.go +++ b/admin.go @@ -241,18 +241,55 @@ func sanitizeBaseURL(raw string) string { // back out through this field, even though adminProviderView.BaseURL // itself is already sanitized. // -// This is best-effort, not a general URL scrubber: it looks only for -// rawBaseURL appearing verbatim inside lastErr and replaces every such -// occurrence with sanitizeBaseURL's cleaned form. An error naming some -// OTHER URL — a redirect target, a differently-formatted variant of the -// same host — is not caught; recognizing every URL shape a wrapped error -// might embed, without risking mangling ordinary error text, is not -// attempted here. +// It scrubs rawBaseURL's credentials in BOTH forms a real error from a +// call against rawBaseURL can carry them (security audit run-1, finding +// F-2): the verbatim URL, and net/http's own password-masked rewrite of +// it. Go's net/http builds a *url.Error via its unexported stripPassword +// (net/http/client.go) whenever the dialed URL's userinfo reports a +// password SET — which net/url's parseAuthority makes true for ANY +// userinfo containing a colon, including the token-as-username form +// "https://TOKEN:@host" whose password is empty. stripPassword rewrites +// "user:pass@" to "user:***@" IN THE URL BEFORE the error string is ever +// built, so for such a URL the error text never contains rawBaseURL +// verbatim; a verbatim-only match then finds nothing to scrub, leaving +// the USERNAME (commonly the token itself) and any query-string +// credential riding the same URL in place. +// +// sanitizeTargetErr (target_health.go) delegates here rather than +// carrying its own copy of this logic: the private copy it used to hold +// is exactly what left this provider-side path unprotected between the +// two changes that introduced them, so the two must never diverge again. +// +// Still best-effort for URLs OTHER than rawBaseURL: an error naming a +// redirect target is not caught. Rejecting a credential-bearing baseUrl +// at construction (providers.go) is the only complete fix; this is the +// containment that changes no behavior. func sanitizeProviderErr(lastErr, rawBaseURL string) string { - if lastErr == "" || rawBaseURL == "" || !strings.Contains(lastErr, rawBaseURL) { + if lastErr == "" || rawBaseURL == "" { return lastErr } - return strings.ReplaceAll(lastErr, rawBaseURL, sanitizeBaseURL(rawBaseURL)) + out := replaceURLForm(lastErr, rawBaseURL) + u, parseErr := url.Parse(rawBaseURL) + if parseErr != nil || u.User == nil { + return out + } + if _, hasPassword := u.User.Password(); !hasPassword { + // No password set means net/http passed the URL through + // unmasked, so the verbatim pass above already matched it. + return out + } + // net/http's exact stripPassword output, mirrored call for call. + masked := strings.Replace(u.String(), u.User.String()+"@", u.User.Username()+":***@", 1) + return replaceURLForm(out, masked) +} + +// replaceURLForm replaces every occurrence of one rendering of a URL in +// msg with sanitizeBaseURL's cleaned form of that same rendering. +func replaceURLForm(msg, form string) string { + if form == "" || !strings.Contains(msg, form) { + return msg + } + return strings.ReplaceAll(msg, form, sanitizeBaseURL(form)) } // adminProviderView is one provider's read-only view in GET diff --git a/failover.go b/failover.go index c04daa7..969826e 100644 --- a/failover.go +++ b/failover.go @@ -76,6 +76,22 @@ type FailoverConfig struct { const ( defaultFailoverMaxAttempts = 3 maxFailoverMaxAttempts = 10 + + // warnComposedOutboundCalls is the composed failover x retry worst + // case above which newGateway logs the product at construction. + // + // Security audit run-1: failover.maxAttempts and retry.attempts are + // each validated against their OWN ceiling and nothing validates their + // PRODUCT, yet they compose multiplicatively at runtime — the + // candidate loop in runMeteredCall wraps retryPolicy.do, so one + // admitted request can issue maxAttempts * (attempts+1) outbound + // provider calls, up to 10 * 4 = 40. Both features default off; with + // both merely switched on at their own defaults the product is + // 3 * 2 = 6. This threshold is that default-on product doubled, so an + // ordinary opt-in stays silent and only a deliberately wide + // configuration is surfaced — at config time, rather than during the + // upstream incident that makes it matter. + warnComposedOutboundCalls = 12 ) // failoverConfig is FailoverConfig, validated and defaulted once at diff --git a/limits.go b/limits.go index 3d295af..d31a1cf 100644 --- a/limits.go +++ b/limits.go @@ -1427,6 +1427,35 @@ func (l *limiter) account(scopes []limitScope, u usage, costMicros int64) { wb := bucketsFor(now) et := enforceTTLsFor() + // Security audit run-1: every figure reaching here is read verbatim + // out of an UPSTREAM response body (provider_openai.go's forwardJSON, + // the translate_*.go stream states, extractPassthroughUsage) into an + // int64, and is written below as an INCRBY DELTA. A negative value + // therefore DECREMENTS this principal's counters — and because + // checkAndCount admits while `used < limit`, a counter driven below + // zero stops every token and cost budget binding for the rest of that + // window. costMicros can also arrive negative from an int64 overflow + // in costMicros' own unguarded multiply (pricing.go) on an absurd + // reported token count. + // + // These counters are increment-only by contract, so a negative delta + // is never legitimate: drop it and say so, loudly enough that a + // misbehaving or hostile upstream is visible rather than silently + // rewriting a budget. Clamped per-component, so one bad field does not + // discard the other two, which may be perfectly valid. + if u.prompt < 0 || u.completion < 0 || costMicros < 0 { + l.logf("%s", fmt.Sprintf("limits: dropping negative usage reported by an upstream (prompt=%d completion=%d costMicros=%d); these counters are increment-only", u.prompt, u.completion, costMicros)) + if u.prompt < 0 { + u.prompt = 0 + } + if u.completion < 0 { + u.completion = 0 + } + if costMicros < 0 { + costMicros = 0 + } + } + entries := make([]counterIncr, 0, len(scopes)*12) for _, sc := range scopes { // A kindModel scope is the one scope kind checkAndCount never @@ -1638,6 +1667,13 @@ const ( // scope can never collide with a user, group, provider or target scope. const kindModel = "model" +// maxModelScopeIDLen bounds the "provider/model" id withModelScope will +// accept as a counter-scope id. Generous next to any real model id — +// the longest in the built-in table is well under 40 bytes — and far +// below the 4MiB an upstream response body could otherwise put there +// (maxAccountingTeeBytes, routes_passthrough.go). Security audit run-1. +const maxModelScopeIDLen = 256 + // withModelScope returns scopes plus a kindModel scope for canonical, the // "provider/model" id that actually SERVED the request. It copies rather // than appending in place: the caller's slice is the same one @@ -1654,11 +1690,59 @@ func withModelScope(scopes []limitScope, canonical string) []limitScope { if canonical == "" || canonical[len(canonical)-1] == '/' { return scopes } + // Security audit run-1: on the passthrough route the model half of + // canonical is read VERBATIM out of the upstream's own response body + // (extractPassthroughUsage, routes_passthrough.go) with no length or + // character-set check, and every distinct id it yields mints its own + // set of counter keys — up to twelve, four of them retained for + // monthWindowTTL (400 days). An upstream (or, where a group sets no + // Models restriction, a caller whose requested id an OpenAI-compatible + // relay echoes back) could therefore grow the counter keyspace without + // bound. Bounding the id here is the one chokepoint every model-scope + // write passes through. + // + // An id that fails these checks drops the MODEL scope only: the + // request still accounts to user, group and total exactly as before, + // which is the same "attribute what we can, invent nothing" rule the + // empty/trailing-slash case above already follows. + if len(canonical) > maxModelScopeIDLen { + return scopes + } + for i := 0; i < len(canonical); i++ { + if canonical[i] < 0x20 || canonical[i] == 0x7f { + return scopes + } + } out := make([]limitScope, len(scopes), len(scopes)+1) copy(out, scopes) return append(out, limitScope{kind: kindModel, id: canonical}) } +// scopesHaveCostBudget reports whether ANY scope in scopes configures a +// money budget — costPerDayUSD or costPerMonthUSD. A scope with nil +// limits (the synthetic total scope, and any user/group that configured +// none) cannot carry one. +// +// Security audit run-1, finding F-1 (high): this is the predicate that +// decides whether serving a model with NO known price would silently +// defeat a control the operator actually configured. It deliberately +// ignores token and request limits — those accrue correctly for an +// unpriced model and keep working; only the cost budget is the one that +// can never fire, because its counter is never written. Matching +// buildBudgetProbes' own `limit <= 0` rule, a zero or negative budget +// means unlimited and so does not count as configured. +func scopesHaveCostBudget(scopes []limitScope) bool { + for _, sc := range scopes { + if sc.limits == nil { + continue + } + if usdToMicros(sc.limits.CostPerDayUSD) > 0 || usdToMicros(sc.limits.CostPerMonthUSD) > 0 { + return true + } + } + return false +} + // metricProvAttempt and metricProvFail are the counter metric names // recordProviderAttempt writes (Feature A, v0.22): every upstream attempt // increments metricProvAttempt; only a provider-fault outcome diff --git a/llmgateway.go b/llmgateway.go index 4c8b290..2083afa 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -140,6 +140,24 @@ type Config struct { // disable the bound entirely with an unbounded number. MaxInFlightBodyRequests int `json:"maxInFlightBodyRequests,omitempty"` PassthroughUnknown bool `json:"passthroughUnknown,omitempty"` + // AllowUnpricedWithCostBudget restores the pre-fix behavior for a + // request whose served model has NO configured price while the calling + // user or group DOES configure costPerDayUSD/costPerMonthUSD. + // + // Security audit run-1, finding F-1 (high): an unpriced model costs 0 + // in accounting, a zero cost writes no cost counter (limiter.account, + // limits.go), and checkAndCount reads exactly that never-written + // counter — so the money budget can never accrue and can never fire. + // The gateway now REFUSES such a request (runMeteredCall, + // routes_unified.go) rather than serving it free against a control the + // operator explicitly configured. + // + // Set true to serve it anyway, accepting that the cost budget does not + // bound that model. This changes nothing for a deployment that + // configures no cost budget at all — the refusal is already scoped to + // callers that set one — and nothing for a model that has a price. + // Prefer adding the missing "pricing" entry over setting this. + AllowUnpricedWithCostBudget bool `json:"allowUnpricedWithCostBudget,omitempty"` } // ProviderConfig describes one upstream LLM provider. @@ -359,6 +377,20 @@ type RedisConfig struct { // an unbounded value would eagerly allocate that many *respConn // structs at construction before a single request arrives. PoolSize int `json:"poolSize,omitempty"` + // TLS dials Redis over TLS instead of plaintext TCP, verifying the + // server's certificate against the host in Address. + // + // Security audit run-1: before this field existed the hand-rolled + // RESP client could only speak plaintext, so the AUTH password, the + // per-principal counter keys and any cached response bodies were + // unprotectable in transit by any configuration — TLS was not merely + // unconfigured, it was impossible. Defaults to false, so an existing + // deployment is unchanged; enable it when the gateway reaches Redis + // over anything the operator does not already encrypt (a service mesh + // with mTLS, an stunnel/TLS sidecar, or a private link all satisfy + // the same invariant externally, in which case leaving this off is + // fine). + TLS bool `json:"tls,omitempty"` } // RetryConfig configures same-provider retry on transient upstream @@ -648,6 +680,21 @@ const maxExplicitBodyAdmissionCap = 10_000 // stays as short as possible. See adminProviderView's identical // convention (admin.go) for the same reasoning spelled out once; this // struct is the other place it applies. +// +// fieldalignment's 192-vs-184 pointer-bytes hint on this struct is a +// DELIBERATE, pre-existing exception, not an oversight — see +// targetHealthLastSweepUnixNano's own doc comment immediately below. +// That field must stay first so sync/atomic's 64-bit operations are +// correctly aligned on armv7 and 386, which Traefik ships builds for, +// and sync/atomic guarantees that alignment only for the first word of +// an allocated struct. Total struct size is unchanged either way (264 +// bytes, verified), so the exception costs no padding; moving +// targetHealthSweeping up to close the gap was tried and made the +// report worse (200 vs 184). Correctness on 32-bit platforms wins over +// the linter hint, so the check is suppressed here rather than the +// field order being "fixed". +// +//nolint:govet type Gateway struct { // targetHealthLastSweepUnixNano MUST stay the first field (F7, // feat/target-health review). It backs maybeSweepTargetHealth's @@ -906,6 +953,22 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str g.failover = failoverCfg g.failoverHealth = newRequestHealthTracker() + // Security audit run-1: nothing validates the PRODUCT of the two + // retry axes, which compose multiplicatively at runtime. Surface it + // once at construction rather than leaving an operator to discover it + // during an upstream incident — the moment the amplification actually + // fires, since retryPolicy.do only retries on 429/5xx. + if failoverCfg.enabled && config.Retry.Enabled { + retries := config.Retry.Attempts + if retries == 0 { + retries = defaultRetryAttempts + } + if worst := failoverCfg.maxAttempts * (retries + 1); worst > warnComposedOutboundCalls { + g.logf("config: failover.maxAttempts (%d) and retry.attempts (%d) compose to as many as %d outbound provider calls for ONE admitted client request; each consumes the provider quota shared by every tenant of this gateway", + failoverCfg.maxAttempts, retries, worst) + } + } + // feat/target-health: validated once, here — passive recording and // exposure stay always-on regardless of this block; Enabled gates // only maybeSweepTargetHealth's own active probe sweep @@ -1064,10 +1127,17 @@ func buildRedisClient(rc *RedisConfig) (*respClient, error) { // PoolSize left at 0 keeps newRESPClient's own self-tuned default // (defaultRespPoolSize, resp.go); an explicit positive value always // overrides it (RedisConfig.PoolSize's own doc comment). + var client *respClient if rc.PoolSize > 0 { - return newRESPClientPool(rc.Address, password, rc.DB, rc.PoolSize), nil + client = newRESPClientPool(rc.Address, password, rc.DB, rc.PoolSize) + } else { + client = newRESPClient(rc.Address, password, rc.DB) } - return newRESPClient(rc.Address, password, rc.DB), nil + // Set on the constructed client rather than threaded through either + // constructor: neither signature needs to change for it, and both + // have many call sites (respClient.tls's own doc comment, resp.go). + client.tls = rc.TLS + return client, nil } // newConfiguredLimiter builds the limiter for config, using client (nil diff --git a/pricing.go b/pricing.go index c14dae1..c4b733d 100644 --- a/pricing.go +++ b/pricing.go @@ -77,10 +77,35 @@ var ( // a price like 8.2 by one micro-USD due to float64 representation), and // each of input/output cost is tokens*priceMicrosPerM/1e6, summed. func costMicros(model string, u usage, overrides map[string]*ModelPricing) int64 { + cost, _ := costMicrosKnown(model, u, overrides) + return cost +} + +// costMicrosKnown is costMicros plus the one bit costMicros throws away: +// whether a price was actually FOUND. An unknown model returns (0, false) +// — a zero that means "cannot be priced", which is NOT the same value a +// genuinely free model produces, even though both are 0. +// +// Security audit run-1, finding F-1 (high): costMicros' single-int64 +// return conflated those two cases, and limiter.account's own +// `if costMicros != 0` guard then skipped every cost-counter write for an +// unpriced model. checkAndCount reads exactly those never-written +// counters, so `used < limit` stayed permanently true and a configured +// costPerDayUSD/costPerMonthUSD could never fire. Any caller gating a +// SPEND control must use this function and handle ok == false; costMicros +// above remains for callers that only need the number (the accounting +// write itself, where a zero is harmless). +// +// This is the distinction resolvedModelMeta.CostKnown (modelmeta.go) +// already draws for the metadata path, and the one candidateCostAllowed +// (failover.go) already relies on to protect real money — its own comment +// calls recording zero for an unpriced model "an accounting shortcut, not +// proof the provider invoices zero". +func costMicrosKnown(model string, u usage, overrides map[string]*ModelPricing) (int64, bool) { price, ok := lookupPricing(model, overrides) if !ok { warnUnknownModel(model) - return 0 + return 0, false } inputMicrosPerM := int64(math.Round(price.InputPerM * tokensPerMillion)) @@ -88,7 +113,7 @@ func costMicros(model string, u usage, overrides map[string]*ModelPricing) int64 inputCost := u.prompt * inputMicrosPerM / tokensPerMillion outputCost := u.completion * outputMicrosPerM / tokensPerMillion - return inputCost + outputCost + return inputCost + outputCost, true } // lookupPricing resolves model's price: overrides first (a nil entry does diff --git a/registry.go b/registry.go index c94b8fc..5d85cef 100644 --- a/registry.go +++ b/registry.go @@ -906,7 +906,14 @@ func (m *modelRegistry) warmFill(ctx context.Context) { cancel() m.finishRefresh(st, m.now(), ids, err) if err != nil { - m.log("%s", fmt.Sprintf("model registry: initial discovery for provider %q failed: %v", name, err)) + // sanitizeProviderErr (admin.go): the error commonly embeds the + // dialed URL verbatim, so a baseUrl an operator misconfigured + // with embedded credentials would otherwise be written to + // stderr in clear — which m.log (g.errorf, logger.go) does + // unconditionally and unfiltered, against its own "Never log + // key material" contract. Security audit run-1, finding F-2. + m.log("%s", fmt.Sprintf("model registry: initial discovery for provider %q failed: %v", + name, sanitizeProviderErr(err.Error(), m.adapters[name].base()))) } // Own timeout budget, review fix (SHOULD-4): fctx above is // spent by listModels — reusing it here would hand the metadata @@ -982,7 +989,12 @@ func (m *modelRegistry) captureModelMetadata(ctx context.Context, name string, a } meta, err := mf.fetchModelMetadata(ctx) if err != nil { - m.log("%s", fmt.Sprintf("model registry: metadata capture for provider %q failed: %v", name, err)) + // Scrubbed for the same reason as warmFill's own discovery-failure + // line: fetchModelMetadata dials adapter.base()+metadataPath, so + // this error carries the identical credential-bearing URL. + // Security audit run-1, finding F-2. + m.log("%s", fmt.Sprintf("model registry: metadata capture for provider %q failed: %v", + name, sanitizeProviderErr(err.Error(), adapter.base()))) return } if meta == nil { @@ -1045,7 +1057,12 @@ func (m *modelRegistry) refreshProvider(name string, st *providerState, adapter } m.finishRefresh(st, m.now(), ids, err) if err != nil { - m.log("%s", fmt.Sprintf("model registry: discovery refresh for provider %q failed: %v", name, err)) + // Scrubbed (security audit run-1, finding F-2). This line + // repeats on EVERY background refresh interval with no dedup, + // so an unreachable provider whose baseUrl embeds a credential + // would otherwise write it to stderr indefinitely. + m.log("%s", fmt.Sprintf("model registry: discovery refresh for provider %q failed: %v", + name, sanitizeProviderErr(err.Error(), adapter.base()))) } }() diff --git a/resp.go b/resp.go index eff67bf..7501924 100644 --- a/resp.go +++ b/resp.go @@ -2,6 +2,7 @@ package traefikllmgateway import ( "bufio" + "crypto/tls" "errors" "fmt" "io" @@ -149,6 +150,23 @@ type respClient struct { addr string password string db int + // tls wraps every dialled connection in TLS when true, verifying the + // server certificate against addr's host. + // + // Security audit run-1: this client could previously ONLY speak + // plaintext TCP — there was no TLS field, no scheme parsing and no + // rediss:// handling anywhere — so the AUTH password, the counter + // keys (which embed principal names verbatim, windowKey in limits.go) + // and whole cached response bodies were unprotectable in transit by + // ANY operator configuration. That is what distinguishes it from an + // operator who merely chose plaintext: the choice was foreclosed. + // + // Off by default, so an existing deployment behaves exactly as + // before; set redis.tls to opt in. Not a constructor parameter + // because newRESPClient/newRESPClientPool have ~50 call sites across + // the test suite and none of them needs it — buildRedisClient + // (llmgateway.go) sets it directly on the constructed client. + tls bool } // newRESPClient returns a respClient for addr, sized by defaultRespPoolSize @@ -383,6 +401,39 @@ func (c *respClient) ensureConnOn(pc *respConn, deadline time.Time) error { if err != nil { return fmt.Errorf("resp: dial %q: %w", c.addr, err) } + + // Optional TLS (security audit run-1, respClient.tls). Wrapping here + // rather than at a separate dial keeps one dial path: everything + // below — SetDeadline, the AUTH/SELECT handshake, and the caller's + // own command round trip — then runs over the TLS stream instead of + // the bare socket, with no other call site needing to know. + // + // ServerName is the host half of the configured address, so the + // certificate is verified against the name the operator configured. + // InsecureSkipVerify is deliberately left false: an opt-in TLS mode + // that silently skipped verification would provide confidentiality + // against a passive observer while giving none against the active + // one, which is the attacker this exists for. + if c.tls { + host, _, splitErr := net.SplitHostPort(c.addr) + if splitErr != nil { + host = c.addr + } + tlsConn := tls.Client(conn, &tls.Config{ + ServerName: host, + MinVersion: tls.VersionTLS12, + }) + if dlErr := tlsConn.SetDeadline(deadline); dlErr != nil { + _ = tlsConn.Close() + return fmt.Errorf("resp: set deadline: %w", dlErr) + } + if hsErr := tlsConn.Handshake(); hsErr != nil { + _ = tlsConn.Close() + return fmt.Errorf("resp: tls handshake %q: %w", c.addr, hsErr) + } + conn = tlsConn + } + pc.conn = conn pc.r = bufio.NewReader(conn) diff --git a/retry.go b/retry.go index 9cfce2f..8a7d183 100644 --- a/retry.go +++ b/retry.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "math/rand" "net/http" "strconv" "time" @@ -32,6 +33,47 @@ const maxRetryAttempts = 3 // Retry-After header — spec §1's "capped at 2s per wait". const maxRetryWait = 2 * time.Second +// maxRetryAfterWait caps a wait this gateway will honour when the +// UPSTREAM itself asked for it via a 429's Retry-After header. It is +// deliberately far larger than maxRetryWait, which caps only the wait +// this gateway computes for itself. +// +// Security audit run-1 (retry/failover fan-out record): previously a +// Retry-After larger than maxRetryWait was DISCARDED and replaced by the +// <=2s exponential ladder, so a saturated provider could not tell this +// gateway to back off for longer than two seconds — the gateway answered +// an explicit overload signal by retrying sooner than asked. Honouring it +// up to this ceiling can only ever make the gateway wait LONGER, never +// sooner, so it strictly reduces outbound pressure. p.waitFn is +// ctx-aware, so a client disconnect or request deadline still aborts the +// wait immediately rather than blocking out the full duration. +const maxRetryAfterWait = 30 * time.Second + +// retryJitterFn spreads a computed backoff so that many requests failing +// at the same instant do not retry in lockstep against the same upstream +// — the synchronized-retry shape that turns one upstream blip into a +// self-sustaining thundering herd (security audit run-1). It applies +// EQUAL jitter: half the computed wait, plus a uniform random draw over +// the other half, so a jittered wait is always at least d/2 and never +// collapses to an immediate retry the way full jitter can. +// +// A package-level var, mirroring retryPolicy.waitFn's own convention, so +// tests can substitute an identity function and keep asserting exact +// durations. It is applied in do, not in waitBefore, so waitBefore stays +// a pure function of (attempt, resp) and its own unit tests keep testing +// the backoff ladder itself rather than the randomness on top of it. +var retryJitterFn = func(d time.Duration) time.Duration { + if d <= 0 { + return d + } + half := d / 2 + // #nosec G404 -- this randomness spreads retry timing to break up a + // synchronized herd; it guards no secret and gates no decision, so a + // predictable draw grants an attacker nothing. crypto/rand would add + // a syscall per retry wait and an error path, for no security gain. + return half + time.Duration(rand.Int63n(int64(d-half)+1)) +} + // retryDrainLimit caps how much of a retried-away response's body // drainAndClose reads before closing it. It is generous enough to let // the underlying connection settle for reuse, but far below @@ -178,9 +220,18 @@ func (p *retryPolicy) waitBefore(attempt int, resp *http.Response) time.Duration if resp != nil && resp.StatusCode == http.StatusTooManyRequests { if ra := resp.Header.Get("Retry-After"); ra != "" { if secs, err := strconv.Atoi(ra); err == nil && secs >= 0 { - if wait := time.Duration(secs) * time.Second; wait <= maxRetryWait { - return wait + // An upstream-requested cool-down is honoured up to + // maxRetryAfterWait, and CLAMPED to it rather than + // discarded, when it asks for longer (security audit + // run-1). Discarding it — the previous behavior — fell + // back to the <=2s ladder and so retried a provider that + // had explicitly asked for a longer pause sooner than it + // asked, which is backwards under overload. + wait := time.Duration(secs) * time.Second + if wait > maxRetryAfterWait { + wait = maxRetryAfterWait } + return wait } } } @@ -239,7 +290,7 @@ func (p *retryPolicy) do(ctx context.Context, call func() (*http.Response, error return resp, err } - wait := p.waitBefore(attempt, resp) + wait := retryJitterFn(p.waitBefore(attempt, resp)) drainAndClose(resp) if !p.waitFn(ctx, wait) { return resp, err diff --git a/retry_test.go b/retry_test.go index 8ea4a1e..526a67e 100644 --- a/retry_test.go +++ b/retry_test.go @@ -45,6 +45,23 @@ func spyWait(waits *[]time.Duration) func(context.Context, time.Duration) bool { } } +// withoutRetryJitter pins retryJitterFn to the identity for one test and +// restores it on cleanup. do() applies jitter ON TOP OF waitBefore's +// ladder (retry.go, security audit run-1), so a test asserting the EXACT +// ladder values neutralises the randomness here rather than loosening its +// assertions into a range — the ladder contract is still pinned exactly, +// and the jitter itself is pinned separately by +// TestRetryPolicy_Do_JitterStaysWithinHalfToFullBackoff. +// +// Safe as a package-var swap because no test in this file calls +// t.Parallel(); if one ever does, it must not use this helper. +func withoutRetryJitter(t *testing.T) { + t.Helper() + prev := retryJitterFn + retryJitterFn = func(d time.Duration) time.Duration { return d } + t.Cleanup(func() { retryJitterFn = prev }) +} + // callSequence returns a call func for retryPolicy.do that returns // results[i] on its i-th invocation (0-indexed), and counts invocations // into calls. @@ -292,6 +309,7 @@ func TestRetryPolicy_Do_EnabledDefaults_CapsAtTwoTotalTries(t *testing.T) { // uses attempts:1 (one retry allowed) and retries exactly once, waits // the expected duration, and returns the second attempt's success. func TestRetryPolicy_Do_RetriesOnTransientThenSucceeds(t *testing.T) { + withoutRetryJitter(t) tests := []struct { name string first func() (*http.Response, error) @@ -341,13 +359,21 @@ func TestRetryPolicy_Do_RetriesOnTransientThenSucceeds(t *testing.T) { } } -// TestRetryPolicy_WaitBefore_RetryAfterHandling pins spec §1's -// Retry-After handling directly: only a 429 whose Retry-After parses as -// a non-negative whole-second count within the 2s cap wins outright. -// Every other shape — over the cap, the HTTP-date form, negative, no +// TestRetryPolicy_WaitBefore_RetryAfterHandling pins Retry-After +// handling directly: a 429 whose Retry-After parses as a non-negative +// whole-second count is HONOURED, clamped to maxRetryAfterWait when it +// asks for longer. Every other shape — the HTTP-date form, negative, no // header at all, or present on a non-429 status — falls back to -// exponential backoff (spec: ">2s -> exponential backoff", not a flat -// 2s), verified here against attempt=1's backoff*2^0 value. +// exponential backoff, verified here against attempt=1's backoff*2^0 +// value. +// +// CONTRACT CHANGE (security audit run-1): a Retry-After above the 2s +// per-wait cap used to be DISCARDED and replaced by that ladder, so a +// saturated provider could not ask this gateway to pause for longer than +// two seconds — it answered an explicit overload signal by retrying +// sooner than asked. maxRetryWait still caps the wait the gateway +// computes for ITSELF; maxRetryAfterWait caps only what the upstream may +// ask for, and clamping can only ever lengthen a wait, never shorten one. func TestRetryPolicy_WaitBefore_RetryAfterHandling(t *testing.T) { p := &retryPolicy{backoff: 100 * time.Millisecond} backoffFallback := 100 * time.Millisecond // attempt=1: backoff * 2^0 @@ -358,7 +384,8 @@ func TestRetryPolicy_WaitBefore_RetryAfterHandling(t *testing.T) { want time.Duration }{ {retryAfterResp(http.StatusTooManyRequests, "1"), "429 with qualifying Retry-After (1s, within cap)", time.Second}, - {retryAfterResp(http.StatusTooManyRequests, "30"), "429 with Retry-After over the 2s cap falls back to exponential backoff", backoffFallback}, + {retryAfterResp(http.StatusTooManyRequests, "30"), "429 with Retry-After over the 2s per-wait cap is still honoured", maxRetryAfterWait}, + {retryAfterResp(http.StatusTooManyRequests, "3600"), "429 with Retry-After beyond maxRetryAfterWait is clamped to it, never discarded", maxRetryAfterWait}, {retryAfterResp(http.StatusTooManyRequests, "Wed, 21 Oct 2015 07:28:00 GMT"), "429 with HTTP-date Retry-After falls back to exponential backoff", backoffFallback}, {retryAfterResp(http.StatusTooManyRequests, "-5"), "429 with negative Retry-After falls back to exponential backoff", backoffFallback}, {fakeResp(http.StatusTooManyRequests, nil), "429 with no Retry-After header falls back to exponential backoff", backoffFallback}, @@ -446,6 +473,7 @@ func TestRetryPolicy_Do_NilPolicy_NoRetry(t *testing.T) { // retries after the first), waiting p.attempts times with exponential // backoff, capped at maxRetryWait. func TestRetryPolicy_Do_AttemptsCap(t *testing.T) { + withoutRetryJitter(t) var calls int32 var waits []time.Duration p := &retryPolicy{enabled: true, attempts: 3, backoff: time.Second, waitFn: spyWait(&waits)} @@ -496,6 +524,51 @@ func TestRetryPolicy_Do_BackoffCappedAtMaxWait(t *testing.T) { } } +// TestRetryPolicy_Do_JitterStaysWithinHalfToFullBackoff pins the jitter +// do() applies on top of waitBefore's ladder (security audit run-1). +// Deterministic backoff made every request that failed at the same +// instant retry in lockstep against the same upstream — the synchronized +// retry shape that turns one upstream blip into a self-sustaining herd. +// Equal jitter keeps each wait within [d/2, d]: never zero, so it can +// never collapse into an immediate retry, and never longer than the +// ladder value it is spreading. +func TestRetryPolicy_Do_JitterStaysWithinHalfToFullBackoff(t *testing.T) { + const backoff = time.Second + seen := make([]time.Duration, 0, 32) + for range 32 { + var calls int32 + var waits []time.Duration + p := &retryPolicy{enabled: true, attempts: 1, backoff: backoff, waitFn: spyWait(&waits)} + call := callSequence(&calls, func() (*http.Response, error) { + return fakeResp(http.StatusServiceUnavailable, nil), nil + }) + + if _, err := p.do(context.Background(), call); err != nil { + t.Fatalf("do: %v", err) + } + if len(waits) != 1 { + t.Fatalf("waits = %v, want exactly 1 (attempts:1)", waits) + } + if waits[0] < backoff/2 || waits[0] > backoff { + t.Fatalf("waits[0] = %v, want within [%v, %v]", waits[0], backoff/2, backoff) + } + seen = append(seen, waits[0]) + } + + // Assert the spread actually exists before concluding jitter works: + // an identity jitter would satisfy every bound check above. + allEqual := true + for _, w := range seen { + if w != seen[0] { + allEqual = false + break + } + } + if allEqual { + t.Errorf("all %d jittered waits were identical (%v); jitter is not spreading them", len(seen), seen[0]) + } +} + // TestRetryPolicy_Do_CtxAlreadyCanceled_NoRetry asserts a context // canceled before do() even starts stops the loop after the first // attempt: the real (ctx-aware) defaultWaitFn resolves the already-closed diff --git a/routes_unified.go b/routes_unified.go index fb540e6..d898bb7 100644 --- a/routes_unified.go +++ b/routes_unified.go @@ -491,6 +491,34 @@ func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scop // for why (F3/F6 fix, adversarial-review). providerName := cand.providerName upstreamModel := cand.upstreamModel + + // Security audit run-1, finding F-1 (high). An unpriced model is + // billed 0, a zero cost writes no cost counter at all + // (limiter.account, limits.go), and checkAndCount then evaluates + // the money budget against that never-written counter — so + // costPerDayUSD/costPerMonthUSD can never accrue and can never + // fire. Serving the request would spend the operator's money + // against a control that provably cannot stop it. + // + // Placed HERE, not in admitRequest, because admitRequest runs + // before the body is decoded (runUnified's own ordering) and so + // before any model id exists; this is the first point where the + // SERVING model is known. It is before call(), so nothing has been + // written to the client yet and the refusal is a clean envelope. + // + // Deliberately narrow: it fires only when some scope actually + // configured a money budget, so a deployment with none is + // unaffected, and token/request budgets are untouched — those + // accrue correctly for an unpriced model and keep working. + if !g.cfg.AllowUnpricedWithCostBudget && + !modelPriceKnown(cand.canonical, cand.upstreamModel, g.cfg.Pricing) && + scopesHaveCostBudget(scopes) { + g.logf("%s: refusing model %q: it has no configured price, and a cost budget applies to this caller that cannot be enforced without one", logPrefix, cand.canonical) + envelope(sw, http.StatusPaymentRequired, "invalid_request_error", + fmt.Sprintf("model %q has no configured price, so the cost budget that applies to this request cannot be enforced; add a \"pricing\" entry for it, or set allowUnpricedWithCostBudget to serve it unbounded", cand.canonical)) + return + } + ctx := withAttemptRecorder(r.Context(), func(resp *http.Response, attemptErr error) { g.limiter.recordProviderAttempt(providerName, upstreamModel, resp, attemptErr) }) @@ -870,10 +898,35 @@ func (g *Gateway) readAndDecodeUnifiedBody(sw *statusTrackingWriter, r *http.Req // computation (and its unknown-model warning, when neither id has a price) // runs through the one costMicros call that is actually charged. func unifiedCostMicros(canonical, bare string, u usage, overrides map[string]*ModelPricing) int64 { + cost, _ := unifiedCostMicrosKnown(canonical, bare, u, overrides) + return cost +} + +// unifiedCostMicrosKnown is unifiedCostMicros plus whether EITHER id +// actually had a price (security audit run-1, finding F-1). The bool is +// false only when neither canonical nor bare is priced — exactly the case +// where the returned 0 means "cannot be priced" rather than "free", and +// therefore the case a spend control must not silently treat as zero +// spend. See costMicrosKnown (pricing.go) for the full reasoning. +func unifiedCostMicrosKnown(canonical, bare string, u usage, overrides map[string]*ModelPricing) (int64, bool) { if _, ok := lookupPricing(canonical, overrides); ok { - return costMicros(canonical, u, overrides) + return costMicrosKnown(canonical, u, overrides) } - return costMicros(bare, u, overrides) + return costMicrosKnown(bare, u, overrides) +} + +// modelPriceKnown reports whether EITHER id this request could be billed +// under has a configured price, using unifiedCostMicrosKnown's own +// canonical-then-bare order. It is a pure existence probe: unlike +// costMicrosKnown it never fires the unknown-model warning, so the +// pre-flight check in runMeteredCall does not double-warn for a request +// that is about to be accounted (and warned about) anyway. +func modelPriceKnown(canonical, bare string, overrides map[string]*ModelPricing) bool { + if _, ok := lookupPricing(canonical, overrides); ok { + return true + } + _, ok := lookupPricing(bare, overrides) + return ok } // writeModelResolveErrorEnvelope maps a modelRegistry.resolve error to diff --git a/sse.go b/sse.go index c826d89..e7806d0 100644 --- a/sse.go +++ b/sse.go @@ -2,6 +2,8 @@ package traefikllmgateway import ( "bufio" + "bytes" + "errors" "io" "net/http" "strings" @@ -18,6 +20,32 @@ const sseScannerInitialBufSize = 64 * 1024 // growing its buffer without bound. const sseScannerMaxLineSize = 4 * 1024 * 1024 +// sseMaxEventBytes caps the TOTAL data bytes readSSE will accumulate for +// ONE event, across however many "data:" lines that event spans. +// +// Security audit run-1: sseScannerMaxLineSize above bounds a single LINE, +// not an event, and readSSE's accumulator resets only on a blank line or +// at EOF — so an upstream that emits many short "data:" lines and never +// terminates the record grows dataLines without any bound at all, and +// peak residency at dispatch is roughly three times that (the joined +// string, its []byte copy, and the slice, all live at once). The per-line +// cap never fires for that shape, because no individual line is large. +// Under deployment form A the heap being grown is the shared Traefik +// ingress process's, so this is not self-inflicted: it is every other +// tenant and every unrelated route on that instance. +// +// The value is generous — an event legitimately carrying a base64 image +// stays far below it — and is a backstop against unbounded growth, not a +// tuning knob. +const sseMaxEventBytes = 8 * 1024 * 1024 + +// errSSEEventTooLarge is returned by readSSE when one event's accumulated +// "data:" payload exceeds sseMaxEventBytes. It mirrors how the per-line +// cap surfaces (bufio.ErrTooLong is returned as-is): the caller's stream +// forwarding stops and the error propagates, rather than the gateway +// continuing to buffer an event that will never terminate. +var errSSEEventTooLarge = errors.New("llmgateway: sse event data exceeded the maximum accumulated size") + // sseDataPrefix is the "data:" field prefix sseWriter writes before every // event payload, per the text/event-stream wire format. const sseDataPrefix = "data: " @@ -66,6 +94,7 @@ func readSSE(r io.Reader, fn func(sseEvent) error) error { event string dataLines []string hasData bool + dataBytes int ) firstLine := true @@ -91,6 +120,7 @@ func readSSE(r io.Reader, fn func(sseEvent) error) error { event = "" dataLines = nil hasData = false + dataBytes = 0 if !fire { return nil } @@ -119,6 +149,13 @@ func readSSE(r io.Reader, fn func(sseEvent) error) error { case "event": event = value case "data": + // +1 for the "\n" this line will contribute to the joined + // payload. Checked BEFORE appending, so the cap bounds what is + // actually held rather than catching it one line late. + dataBytes += len(value) + 1 + if dataBytes > sseMaxEventBytes { + return errSSEEventTooLarge + } dataLines = append(dataLines, value) hasData = true case "id", "retry": @@ -175,17 +212,34 @@ func newSSEWriter(w http.ResponseWriter) *sseWriter { return &sseWriter{w: w, f: f} } -// writeData writes one SSE data event, "data: " + b + "\n\n", as a -// single Write call. It flushes when the underlying writer supports -// that. b must not contain a newline. The text/event-stream format reads -// everything up to the first newline as the field value, so a raw -// newline inside b splits the frame into two malformed lines. Callers -// must pass compact JSON, not pretty-printed JSON, as b. +// writeData writes one SSE data event as a single Write call, flushing +// when the underlying writer supports that. +// +// A newline inside b is now ENCODED rather than emitted raw: b is split +// on "\n" and each line gets its own "data: " prefix, which is exactly +// how the text/event-stream format represents a multi-line payload, and +// is what a conforming client rejoins with "\n" on the other side. For +// single-line b — every caller that passes compact JSON, which is still +// what callers should pass — the bytes written are byte-for-byte +// identical to before. +// +// Security audit run-1: this function previously DOCUMENTED that b must +// not contain a newline and enforced nothing, while readSSE on the other +// side is specified (and unit-tested) to JOIN multi-line data with "\n". +// openaiAdapter.forwardStream relays readSSE's output straight here, so +// an upstream could close the frame early and inject a complete second +// event — carrying its own event:/id:/retry: fields, which this gateway +// otherwise strips — into the client's stream. Encoding the newline +// closes that by construction, with no validation branch to get wrong. func (s *sseWriter) writeData(b []byte) error { - buf := make([]byte, 0, len(sseDataPrefix)+len(b)+2) - buf = append(buf, sseDataPrefix...) - buf = append(buf, b...) - buf = append(buf, '\n', '\n') + lines := bytes.Split(b, []byte{'\n'}) + buf := make([]byte, 0, len(b)+len(lines)*len(sseDataPrefix)+1) + for _, line := range lines { + buf = append(buf, sseDataPrefix...) + buf = append(buf, line...) + buf = append(buf, '\n') + } + buf = append(buf, '\n') if _, err := s.w.Write(buf); err != nil { return err diff --git a/target_health.go b/target_health.go index 07499a8..1bc4bde 100644 --- a/target_health.go +++ b/target_health.go @@ -5,8 +5,6 @@ import ( "errors" "fmt" "net/http" - "net/url" - "strings" "sync" "sync/atomic" "time" @@ -208,17 +206,14 @@ func (t *targetHealthTracker) record(kind, name, rawURL string, ok bool, err err // carries a password — builds net/http's exact masked form itself // (mirroring stripPassword's own strings.Replace call precisely) and // scrubs against that too. +// (Security audit run-1, finding F-2): the URL-aware second pass this +// function used to carry privately now lives in sanitizeProviderErr +// (admin.go), so the target path and the provider path share ONE +// implementation. That private copy is precisely why the provider path +// went unprotected — this delegation is what stops the two diverging +// again. func sanitizeTargetErr(msg, rawURL string) string { - msg = sanitizeProviderErr(msg, rawURL) - u, parseErr := url.Parse(rawURL) - if parseErr != nil || u.User == nil { - return msg - } - if _, hasPassword := u.User.Password(); !hasPassword { - return msg - } - masked := strings.Replace(u.String(), u.User.String()+"@", u.User.Username()+":***@", 1) - return sanitizeProviderErr(msg, masked) + return sanitizeProviderErr(msg, rawURL) } // stateOf reports (kind, name)'s current targetHealthState — see this