traefik-llmgateway/routes_passthrough.go

package traefikllmgateway

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/url"
	"strings"
)

// maxPassthroughBytes caps a client passthrough request body: 32MiB,
// generous enough for a native multimodal payload (audio, image, or a
// large document) that the unified routes' 10MiB maxRequestBytes does not
// need to accommodate, while still bounding memory against an oversized
// or malicious body.
const maxPassthroughBytes = 32 << 20

// maxAccountingTeeBytes caps how much of a non-streaming JSON response
// body handlePassthrough tees off for best-effort usage accounting: 4MiB,
// generous for a usage object even inside a very large chat response,
// while the client-facing io.Copy streams every byte of the response —
// of any size — without ever waiting for the full body to buffer first.
const maxAccountingTeeBytes = 4 << 20

// hopByHopHeaders lists the RFC 7230 §6.1 hop-by-hop headers, stripped
// from both the outgoing upstream request and the response copied back
// to the client — they describe this one TCP hop, not the end-to-end
// message, and must never be relayed by an intermediary.
var hopByHopHeaders = map[string]bool{
	"Connection":          true,
	"Keep-Alive":          true,
	"Transfer-Encoding":   true,
	"Upgrade":             true,
	"Proxy-Authorization": true,
	"Proxy-Connection":    true,
	"Te":                  true,
	"Trailer":             true,
}

// gatewayCredentialHeaders are the client's own gateway-authentication
// headers — auth.go's presentedKey reads exactly these two. Stripped from
// the outgoing upstream request so a client's gateway API key never
// reaches the upstream provider; adapter.injectAuth sets the provider's
// own credential afterward.
var gatewayCredentialHeaders = map[string]bool{
	"Authorization": true,
	"X-Api-Key":     true,
}

// clientNegotiationHeaders are content-negotiation headers stripped from
// the outgoing upstream request only, never from the response:
// Accept-Encoding, so Go's http.Transport negotiates compression itself
// and transparently decompresses the reply — per its documented contract,
// that automatic decompression only happens when Transport itself added
// the header, not when the caller (this gateway, forwarding the client's
// own Accept-Encoding) set one explicitly. Without this strip, a client
// that sends "Accept-Encoding: gzip" (every mainstream OpenAI SDK and
// curl --compressed does) gets its exact header forwarded verbatim,
// Transport hands back raw gzip bytes, and the accounting JSON parse in
// handlePassthrough silently fails on binary it cannot decode.
var clientNegotiationHeaders = map[string]bool{
	"Accept-Encoding": true,
}

// copyHeadersExcept copies every header in src to dst, skipping any key
// present in any of excepts. Header keys from both an *http.Request
// parsed off the wire and an *http.Response parsed by the client's
// Transport are already canonicalized by net/http, so a plain map lookup
// against the canonical names above is exact.
func copyHeadersExcept(dst, src http.Header, excepts ...map[string]bool) {
	for k, vs := range src {
		skip := false
		for _, ex := range excepts {
			if ex[k] {
				skip = true
				break
			}
		}
		if skip {
			continue
		}
		dst[k] = append([]string(nil), vs...)
	}
}

// passthroughRoute splits path into a candidate provider name and the
// remaining rest path, per the native passthrough convention
// "/{providerName}/{rest...}". Callers must pass path as
// r.URL.EscapedPath(), never the decoded r.URL.Path: a decoded path
// collapses a client's percent-encoded "%2f" into a literal "/" before it
// ever reaches this split, silently turning what the client sent as one
// opaque path segment into extra routing segments (and, downstream, into
// a "#" that would truncate the rest of the upstream URL at a fragment
// boundary if the encoded byte were "%23"). ok is false for an empty or
// "/"-only path, which carries no candidate provider name at all — the
// caller falls through to its next-route/404 handling in that case.
// passthroughRoute makes no claim the returned providerName is actually
// configured; the caller checks that against g.adapters before treating
// the request as passthrough.
func passthroughRoute(path string) (providerName, rest string, ok bool) {
	trimmed := strings.TrimPrefix(path, "/")
	if trimmed == "" {
		return "", "", false
	}
	providerName, rest, _ = strings.Cut(trimmed, "/")
	if providerName == "" {
		return "", "", false
	}
	return providerName, rest, true
}

// hasTraversalSegment reports whether rest, once percent-decoded, contains
// a path segment exactly equal to "..". rest is still in its escaped form
// here (see passthroughRoute) — decoding it is a validation-only step,
// never used to build the outgoing upstream URL, so a legitimate
// percent-encoded segment (e.g. "a%2Fb" naming a literal "a/b" resource)
// still reaches the upstream exactly as the client sent it. Checking the
// decoded form, not the escaped one, is what catches an encoded traversal
// attempt like "..%2f..%2fsecret": as a single escaped segment it has no
// literal "/" for a naive split to catch, but a permissive upstream
// server decoding that same "%2f" itself would read it as "../../secret"
// — this check rejects it here instead. A rest that fails to
// url.PathUnescape at all is rejected too: a malformed percent-encoding
// is not a path this gateway can reason about safely.
func hasTraversalSegment(rest string) bool {
	decoded, err := url.PathUnescape(rest)
	if err != nil {
		return true
	}
	for _, seg := range strings.Split(decoded, "/") {
		if seg == ".." {
			return true
		}
	}
	return false
}

// cappedAccountingBuffer is an io.Writer that accumulates up to
// maxAccountingTeeBytes of a response body, then silently discards the
// rest while still reporting every byte as written. It is the accounting
// side of an io.TeeReader wrapped around a non-streaming JSON response
// body in handlePassthrough: the client-facing io.Copy reads the real
// response (of any size) through the tee, and this buffer only ever
// retains the first maxAccountingTeeBytes of it for a best-effort usage
// parse afterward. truncated records whether the cap was hit, so the
// caller knows a parse would only fail on cut-off JSON and skips it.
type cappedAccountingBuffer struct {
	buf       bytes.Buffer
	truncated bool
}

// Write implements io.Writer. It always reports success for the full
// input length — even once truncated is set and further bytes are being
// discarded — because this type is driven by io.TeeReader, whose Read
// aborts the underlying copy with an error the moment a tee Write returns
// anything short of len(p); silently dropping bytes past the cap must
// never disrupt the real, client-facing copy this buffer is only
// observing.
func (c *cappedAccountingBuffer) Write(p []byte) (int, error) {
	n := len(p)
	if c.truncated {
		return n, nil
	}
	remaining := maxAccountingTeeBytes - c.buf.Len()
	if remaining <= 0 {
		c.truncated = true
		return n, nil
	}
	if len(p) > remaining {
		p = p[:remaining]
		c.truncated = true
	}
	c.buf.Write(p) //nolint:errcheck // bytes.Buffer.Write never errors
	return n, nil
}

// passthroughUsagePayload captures every shape the three provider types'
// native non-streaming JSON responses report token usage in, plus the
// optional top-level "model" field OpenAI- and Anthropic-shaped responses
// carry (Gemini's native response carries neither "model" nor any of the
// URL-supplied model id).
type passthroughUsagePayload struct {
	Model string `json:"model"`
	Usage struct {
		PromptTokens     int64 `json:"prompt_tokens"`
		CompletionTokens int64 `json:"completion_tokens"`
		InputTokens      int64 `json:"input_tokens"`
		OutputTokens     int64 `json:"output_tokens"`
	} `json:"usage"`
	UsageMetadata struct {
		PromptTokenCount     int64 `json:"promptTokenCount"`
		CandidatesTokenCount int64 `json:"candidatesTokenCount"`
	} `json:"usageMetadata"`
}

// extractPassthroughUsage best-effort parses body as one of the three
// provider types' native non-streaming JSON response shapes, returning
// its reported token usage and model id, keyed by typeName (one of
// providerTypeOpenAI/Anthropic/Gemini). The returned error is body's own
// json.Unmarshal error, when decoding failed — the caller logs it rather
// than discarding it, since a body that should have been JSON (its
// Content-Type said so) but did not decode as JSON is worth knowing
// about, not silently swallowing. Usage and model still come back as
// their best-effort zero/fallback values in that case: a zero usage is a
// no-op for the caller's account call, not a wrong charge. The model id
// falls back to "unknown/"+providerName when the response carries no
// "model" field of its own, which is always true for Gemini's native
// response.
func extractPassthroughUsage(typeName, providerName string, body []byte) (usage, string, error) {
	var payload passthroughUsagePayload
	err := json.Unmarshal(body, &payload)

	model := payload.Model
	if model == "" {
		model = "unknown/" + providerName
	}

	switch typeName {
	case providerTypeOpenAI:
		return usage{prompt: payload.Usage.PromptTokens, completion: payload.Usage.CompletionTokens}, model, err
	case providerTypeAnthropic:
		return usage{prompt: payload.Usage.InputTokens, completion: payload.Usage.OutputTokens}, model, err
	case providerTypeGemini:
		return usage{prompt: payload.UsageMetadata.PromptTokenCount, completion: payload.UsageMetadata.CandidatesTokenCount}, model, err
	default:
		return usage{}, model, err
	}
}

// handlePassthrough implements the native provider passthrough route:
// "/{providerName}/{rest...}" reverse-proxies rest, verbatim, to
// providerName's configured upstream, with the client's gateway
// credential swapped for the provider's own. It is a raw proxy, not a
// translation layer like the unified routes — a non-2xx upstream status
// is forwarded to the client exactly as the upstream sent it, not wrapped
// in the gateway's own error envelope the way the unified routes wrap a
// providerHTTPError.
//
// Checks run in this order: group authorization (403) before capability
// checks (Upgrade→501, path validity→400) before rate limits (429/503) —
// a caller who cannot use providerName at all learns that first, rather
// than learning something about how they tried to use it.
//
// For a Gemini provider, rest is appended to base() exactly as the client
// sent it: there is no model extraction or URL rewriting here, so a
// Gemini passthrough client must address it with Gemini's own native URL
// structure, including its "/v1beta/models/{model}:generateContent"
// paths — injectAuth still sets the same x-goog-api-key header it sets
// for every other Gemini request.
func (g *Gateway) handlePassthrough(w http.ResponseWriter, r *http.Request, u *user, grp *group, providerName, rest string) {
	if !grp.allowsProvider(providerName) {
		writeOAIError(w, http.StatusForbidden, "invalid_request_error", "provider access denied")
		return
	}

	if r.Header.Get("Upgrade") != "" {
		writeOAIError(w, http.StatusNotImplemented, "invalid_request_error", "websocket/upgrade passthrough not supported")
		return
	}

	if hasTraversalSegment(rest) {
		writeOAIError(w, http.StatusBadRequest, "invalid_request_error", "invalid path")
		return
	}

	scopes := withTotalScope(buildLimitScopes(u, grp))
	if violation := g.limiter.checkAndCount(scopes); violation != nil {
		writeLimitViolation(w, violation)
		return
	}

	// providerName is only ever passed here by ServeHTTP's dispatch, which
	// already confirmed it is a key of g.adapters before calling in.
	adapter := g.adapters[providerName]

	upstreamURL := adapter.base() + "/" + rest
	if r.URL.RawQuery != "" {
		upstreamURL += "?" + r.URL.RawQuery
	}

	// Feature A (v0.22): provider-level only — unlike runUnified/the media
	// routes, the upstream model here lives in the response body
	// (extractPassthroughUsage, below), read only after this attempt
	// already resolved, so there is no model to attribute it to yet. See
	// recordProviderAttempt's own doc comment (limits.go).
	r = r.WithContext(withAttemptRecorder(r.Context(), func(resp *http.Response, attemptErr error) {
		g.limiter.recordProviderAttempt(providerName, "", resp, attemptErr)
	}))

	result, ok := g.proxyUpstream(w, r, upstreamURL, adapter.httpClient(), adapter.injectAuth, true, "passthrough (provider "+providerName+")")
	if !ok || !result.isJSON {
		// A build/connection/copy failure already wrote its own response
		// (or, for a canceled client context, wrote nothing at all — see
		// proxyUpstream); a non-JSON response has nothing more to account
		// than the request itself, already counted by checkAndCount above.
		return
	}
	if result.tee.truncated {
		g.logf("passthrough: response body exceeded %d bytes; skipping usage accounting (provider %q)", maxAccountingTeeBytes, providerName)
		return
	}

	respUsage, model, unmarshalErr := extractPassthroughUsage(adapter.typeName(), providerName, result.tee.buf.Bytes())
	if unmarshalErr != nil {
		g.logf("passthrough: response body did not decode as JSON for usage accounting (provider %q): %v", providerName, unmarshalErr)
	}
	cost := unifiedCostMicros(providerName+"/"+model, model, respUsage, g.cfg.Pricing)
	g.limiter.account(scopes, respUsage, cost)
}

// proxyResult is what proxyUpstream reports back to its caller once it has
// streamed a response to the client: enough for a caller that wants
// best-effort JSON usage accounting (handlePassthrough) to run it, without
// proxyUpstream itself knowing anything about usage or pricing. tee is
// nil unless accountJSON was true and the response's Content-Type was
// application/json.
type proxyResult struct {
	tee    *cappedAccountingBuffer
	isJSON bool
}

// proxyUpstream is the shared reverse-proxy core behind both native
// provider passthrough (handlePassthrough, above) and the MCP/A2A target
// proxy (handleTargetProxy, mcp_a2a.go): it builds an upstream request
// from r's method/body/headers, sends it over client, and streams the
// response back to w incrementally through a flushWriter — so an SSE or
// other chunked upstream body reaches the client as each chunk arrives,
// never buffered until the copy finishes.
//
// Every hop-by-hop header (hopByHopHeaders) and the client's own gateway
// credential (gatewayCredentialHeaders) are stripped from the outgoing
// request, and Accept-Encoding (clientNegotiationHeaders) besides, so
// Transport can negotiate and transparently decompress compression
// itself. injectAuth, when non-nil, is called on the built request before
// it is sent — handlePassthrough passes its adapter's injectAuth to swap
// the client's key for the provider's own; handleTargetProxy passes nil,
// since an MCP server or A2A agent is an in-cluster target that receives
// no injected credential at all.
//
// A build failure or a dead upstream writes a 502 envelope to w and
// returns ok=false; a canceled client context (errors.Is context.Canceled)
// is logged, not surfaced, and also returns ok=false, writing nothing —
// the client is already gone. A response-copy failure after headers were
// already written also returns ok=false, since nothing further can be
// written to w at that point either way. logPrefix labels every
// logf/errorf line this call emits, so passthrough and target-proxy
// failures stay distinguishable in the log.
//
// accountJSON gates whether a non-streaming application/json response
// gets teed off into result.tee for the caller's own best-effort usage
// parse afterward: handlePassthrough passes true; handleTargetProxy
// passes false, since target-proxy accounting never goes past the
// request-count checkAndCount already ran before calling in, and teeing a
// response nobody will ever read back would only cost memory for nothing.
func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstreamURL string, client *http.Client, injectAuth func(*http.Request), accountJSON bool, logPrefix string) (result proxyResult, ok bool) {
	bodyReader := io.LimitReader(r.Body, maxPassthroughBytes)
	// gosec G704 (SSRF via taint analysis) flags upstreamURL as
	// request-derived: it is, by design — this is a reverse proxy, and its
	// whole job is to forward a client-supplied path/query onto the
	// upstream. The host component is never request-derived: every caller
	// builds upstreamURL as an operator-configured base (a provider's
	// adapter.base(), or an MCP-server/agent TargetConfig.URL validated at
	// construction — see validateTargetURLs) plus "/"+rest — string
	// concatenation, not a second url.Parse of rest alone — so no
	// client-supplied value can change the scheme or host url.Parse
	// resolves for the final string. rest is additionally
	// traversal-checked by every caller before this point is reached
	// (hasTraversalSegment rejects any ".." segment in its decoded form)
	// and is forwarded here in its original escaped form — see
	// passthroughRoute's and targetRoute's EscapedPath()-based splits — so
	// a client cannot smuggle an encoded "/" (%2f) past this gateway's own
	// routing only to have a permissive upstream reinterpret it as a real
	// separator.
	upstreamReq, err := http.NewRequestWithContext(r.Context(), r.Method, upstreamURL, bodyReader) //nolint:gosec // operator-fixed host, rest is traversal-checked and forwarded escaped; see comment above
	if err != nil {
		g.errorf("%s: build upstream request: %v", logPrefix, err)
		writeOAIError(w, http.StatusBadGateway, "server_error", "upstream connection error")
		return proxyResult{}, false
	}
	upstreamReq.ContentLength = r.ContentLength
	if r.ContentLength < 0 || r.ContentLength > maxPassthroughBytes {
		// Unknown or over-cap length: let the transport negotiate chunked
		// transfer instead of advertising a Content-Length the capped
		// bodyReader above may not actually deliver.
		upstreamReq.ContentLength = -1
	}
	copyHeadersExcept(upstreamReq.Header, r.Header, hopByHopHeaders, gatewayCredentialHeaders, clientNegotiationHeaders)
	if injectAuth != nil {
		injectAuth(upstreamReq)
	}

	resp, err := client.Do(upstreamReq) //nolint:gosec // same upstreamReq built above; operator-fixed host, traversal-checked, see its construction comment
	// Feature A (v0.22): proxyUpstream makes exactly one attempt (no
	// retry.go policy wraps this path), so this fires once per call,
	// whichever way it resolves. Only handlePassthrough's own context ever
	// carries a recorder (attemptRecorderFromContext, providers.go) —
	// handleTargetProxy (mcp_a2a.go), this function's other caller, never
	// wraps r's context this way, so an MCP/A2A target proxy attempt is
	// correctly never accounted as provider traffic.
	if rec := attemptRecorderFromContext(r.Context()); rec != nil {
		rec(resp, err)
	}
	if err != nil {
		if errors.Is(err, context.Canceled) {
			g.logf("%s: client canceled request: %v", logPrefix, err)
			return proxyResult{}, false
		}
		g.errorf("%s: upstream connection error: %v", logPrefix, err)
		writeOAIError(w, http.StatusBadGateway, "server_error", "upstream connection error")
		return proxyResult{}, false
	}
	defer resp.Body.Close() //nolint:errcheck // read-side close; nothing actionable on failure

	copyHeadersExcept(w.Header(), resp.Header, hopByHopHeaders)
	w.WriteHeader(resp.StatusCode)
	fw := newFlushWriter(w)

	// Only a non-streaming JSON response gets teed off for best-effort
	// accounting, and only when the caller wants that at all (accountJSON);
	// every other response streams straight through fw with no extra
	// buffering — a large or slow upstream body must never sit waiting for
	// full receipt before the client sees its first byte.
	isJSON := accountJSON && strings.Contains(strings.ToLower(resp.Header.Get("Content-Type")), "application/json")
	var tee *cappedAccountingBuffer
	var reader io.Reader = resp.Body
	if isJSON {
		tee = &cappedAccountingBuffer{}
		reader = io.TeeReader(resp.Body, tee)
	}

	if _, err := io.Copy(fw, reader); err != nil {
		g.errorf("%s: stream response body: %v", logPrefix, err)
		return proxyResult{isJSON: isJSON, tee: tee}, false
	}
	return proxyResult{isJSON: isJSON, tee: tee}, true
}