mirror of
https://github.com/QuantumNous/new-api.git
synced 2026-10-28 08:35:33 +08:00
fix(relay): mask only the custom upstream address in error messages (#7822)
* fix(relay): mask only the upstream address in error messages MaskSensitiveInfo masked every URL, domain and IPv4 in upstream error messages, which garbled documentation links, email domains and dotted identifiers that clients need to read. It now takes the upstream addresses of the request: the channel base URL, the origin the gateway actually dialed (recorded in doRequest, DoWssRequest, the AWS SDK client and the Xunfei and Volcengine TTS dials, which covers absolute Advanced Custom routes), the channel proxy and the environment proxy. URLs and bare hostnames on those hosts and all IPv4 addresses are masked; other URLs and domains stay readable. With no upstream address no address is masked. Credentials are masked on every URL regardless of host: api_key values, query values and userinfo. The URL matcher also keeps quotes, colons and semicolons that follow a URL, keeps parentheses and apostrophes inside it, masks an IPv4 URL host whole and keeps a country-code TLD when the host has a port. * fix(relay): keep preset official endpoints readable in masked errors A channel on the base URL new-api presets for its type reaches a public official endpoint, so its address is not masked. Only addresses the admin configured are: a custom base URL, the origin of an Advanced Custom route, and the channel and environment proxies. Preset hosts are compared with their port so a configured localhost upstream stays apart from the Ollama preset. The origins recorded for the AWS SDK, Xunfei and Volcengine TTS dials and the plugin allowedHosts masking are dropped: they are all official endpoints. * fix(relay): mask IPv4 addresses and plugin allowed hosts in every error IPv4 addresses in network errors include the gateway's own address, so they are masked even when the channel uses a preset base URL. Task plugin build failures mask the plugin's allowedHosts again: an installed plugin may point them at private hosts. * fix(relay): mask bracketed IPv6 addresses in errors Go network errors write IPv6 addresses in brackets, as in "dial tcp [2001:db8::1]:443". Like IPv4, they are the resolved upstream or the gateway's own address, so they are masked on every error once the bracketed content parses as an IP.
This commit is contained in:
1 parent
718a4e7f97
commit
19d435e7fd
23 files changed
+550
-96
No files matched your search
@@ -10,12 +10,15 @@ import (
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/QuantumNous/new-api/constant"
|
||||
"github.com/QuantumNous/new-api/relaykit/dto"
|
||||
"github.com/pkg/errors"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"golang.org/x/net/http/httpproxy"
|
||||
)
|
||||
|
||||
const KeyRequestBody = "key_request_body"
|
||||
@@ -175,6 +178,62 @@ func GetContextKeyString(c *gin.Context, key constant.ContextKey) string {
|
||||
return c.GetString(string(key))
|
||||
}
|
||||
|
||||
// presetUpstreamHosts are the hosts, with any port, of the default base URLs
|
||||
// new-api presets for channel types. They are public endpoints, so errors keep
|
||||
// them readable. The port keeps a configured localhost upstream apart from the
|
||||
// Ollama preset.
|
||||
var presetUpstreamHosts = sync.OnceValue(func() map[string]struct{} {
|
||||
hosts := make(map[string]struct{}, len(constant.ChannelBaseURLs))
|
||||
for _, baseURL := range constant.ChannelBaseURLs {
|
||||
if parsed, err := url.Parse(baseURL); err == nil && parsed.Host != "" {
|
||||
hosts[strings.ToLower(parsed.Host)] = struct{}{}
|
||||
}
|
||||
}
|
||||
return hosts
|
||||
})
|
||||
|
||||
// isPresetUpstreamHost reports whether address points at one of the preset
|
||||
// default base URL hosts.
|
||||
func isPresetUpstreamHost(address string) bool {
|
||||
rawURL := strings.TrimSpace(address)
|
||||
if !strings.Contains(rawURL, "://") {
|
||||
rawURL = "//" + rawURL
|
||||
}
|
||||
parsed, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
_, preset := presetUpstreamHosts()[strings.ToLower(parsed.Host)]
|
||||
return preset
|
||||
}
|
||||
|
||||
// UpstreamAddresses returns the addresses masked in errors shown to users: the
|
||||
// upstream the admin configured for the selected channel and the proxies its
|
||||
// requests may go through. Preset default base URLs are public and stay
|
||||
// readable. The origin actually requested only counts for a configured base
|
||||
// URL or an Advanced Custom channel, whose routes may target absolute URLs;
|
||||
// other adaptors build official endpoints themselves.
|
||||
func UpstreamAddresses(c *gin.Context) []string {
|
||||
// A channel without its own proxy dials through the environment proxy.
|
||||
environmentProxy := httpproxy.FromEnvironment()
|
||||
candidates := []string{environmentProxy.HTTPProxy, environmentProxy.HTTPSProxy}
|
||||
if setting, ok := GetContextKeyType[dto.ChannelSettings](c, constant.ContextKeyChannelSetting); ok {
|
||||
candidates = append(candidates, setting.Proxy)
|
||||
}
|
||||
baseURL := strings.TrimSpace(GetContextKeyString(c, constant.ContextKeyChannelBaseUrl))
|
||||
candidates = append(candidates, baseURL)
|
||||
if GetContextKeyInt(c, constant.ContextKeyChannelType) == constant.ChannelTypeAdvancedCustom || (baseURL != "" && !isPresetUpstreamHost(baseURL)) {
|
||||
candidates = append(candidates, GetContextKeyString(c, constant.ContextKeyUpstreamRequestOrigin))
|
||||
}
|
||||
addresses := make([]string, 0, len(candidates))
|
||||
for _, address := range candidates {
|
||||
if strings.TrimSpace(address) != "" && !isPresetUpstreamHost(address) {
|
||||
addresses = append(addresses, address)
|
||||
}
|
||||
}
|
||||
return addresses
|
||||
}
|
||||
|
||||
func GetContextKeyInt(c *gin.Context, key constant.ContextKey) int {
|
||||
return c.GetInt(string(key))
|
||||
}
|
||||
|
||||
+2
-2
@@ -135,6 +135,6 @@ func MaskEmail(email string) string {
|
||||
|
||||
// MaskSensitiveInfo moved to the conversion kit (kitutil) because the types
|
||||
// package error formatting depends on it; host callers keep this name.
|
||||
func MaskSensitiveInfo(str string) string {
|
||||
return kitutil.MaskSensitiveInfo(str)
|
||||
func MaskSensitiveInfo(str string, upstreamURLs ...string) string {
|
||||
return kitutil.MaskSensitiveInfo(str, upstreamURLs...)
|
||||
}
|
||||
@@ -40,6 +40,11 @@ const (
|
||||
ContextKeyChannelMultiKeyIndex ContextKey = "channel_multi_key_index"
|
||||
ContextKeyChannelKey ContextKey = "channel_key"
|
||||
|
||||
// ContextKeyUpstreamRequestOrigin is the scheme and host of the last upstream
|
||||
// request, which differs from the base URL for absolute Advanced Custom routes
|
||||
// and for adaptors that build official endpoints themselves.
|
||||
ContextKeyUpstreamRequestOrigin ContextKey = "upstream_request_origin"
|
||||
|
||||
ContextKeyAutoGroup ContextKey = "auto_group"
|
||||
ContextKeyAutoGroupIndex ContextKey = "auto_group_index"
|
||||
ContextKeyAutoGroupRetryIndex ContextKey = "auto_group_retry_index"
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/QuantumNous/new-api/common"
|
||||
"github.com/QuantumNous/new-api/i18n"
|
||||
"github.com/QuantumNous/new-api/middleware"
|
||||
"github.com/QuantumNous/new-api/model"
|
||||
@@ -19,7 +20,7 @@ func Playground(c *gin.Context) {
|
||||
defer func() {
|
||||
if newAPIError != nil {
|
||||
c.JSON(newAPIError.StatusCode, gin.H{
|
||||
"error": newAPIError.ToOpenAIError(),
|
||||
"error": newAPIError.ToOpenAIError(common.UpstreamAddresses(c)...),
|
||||
})
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -107,7 +107,10 @@ func serveTaskPluginImageProtocol(c *gin.Context, pinned pluginruntime.PinnedEnd
|
||||
if reason == "" {
|
||||
reason = "image generation failed"
|
||||
}
|
||||
respondTaskPluginImageError(c, service.TaskErrorWrapperLocal(errors.New(reason), "image_generation_failed", http.StatusBadRequest))
|
||||
// The fail reason comes from the upstream, so mask the channel address in it.
|
||||
taskErr = service.TaskErrorWrapper(errors.New(reason), "image_generation_failed", http.StatusBadRequest, common.UpstreamAddresses(c)...)
|
||||
taskErr.LocalError = true
|
||||
respondTaskPluginImageError(c, taskErr)
|
||||
return
|
||||
}
|
||||
view, err := service.BuildTaskPluginView(task)
|
||||
@@ -156,7 +159,7 @@ func serveTaskPluginImageProtocol(c *gin.Context, pinned pluginruntime.PinnedEnd
|
||||
}
|
||||
_, encoded, downloadErr := deps.downloadImage(url)
|
||||
if downloadErr != nil {
|
||||
logger.LogWarn(c, common.LogText("task protocol image download failed; returning the upstream URL instead of b64_json: %s", common.MaskSensitiveInfo(downloadErr.Error())))
|
||||
logger.LogWarn(c, common.LogText("task protocol image download failed; returning the upstream URL instead of b64_json: %s", common.MaskSensitiveInfo(downloadErr.Error(), common.UpstreamAddresses(c)...)))
|
||||
continue
|
||||
}
|
||||
item["b64_json"] = encoded
|
||||
@@ -185,7 +188,7 @@ func waitTaskPluginImageTask(c *gin.Context, task *model.Task, deps pluginProtoc
|
||||
case <-timer.C:
|
||||
}
|
||||
if err := deps.pollTask(ctx, task); err != nil {
|
||||
logger.LogWarn(c, common.LogText("task protocol image poll failed for task %s: %s", task.TaskID, common.MaskSensitiveInfo(err.Error())))
|
||||
logger.LogWarn(c, common.LogText("task protocol image poll failed for task %s: %s", task.TaskID, common.MaskSensitiveInfo(err.Error(), common.UpstreamAddresses(c)...)))
|
||||
continue
|
||||
}
|
||||
if task.Status == model.TaskStatusSuccess || task.Status == model.TaskStatusFailure {
|
||||
|
||||
@@ -1731,6 +1731,7 @@ func TestServeTaskPluginImageProtocolWaitsForAsynchronousTask(t *testing.T) {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
pinned := imageProtocolTestEndpoint(t)
|
||||
c, recorder := newImageProtocolTestContext("")
|
||||
common.SetContextKey(c, constant.ContextKeyChannelBaseUrl, "https://private.example.com")
|
||||
deps := pluginProtocolTestDeps()
|
||||
deps.imagePollInterval = time.Millisecond
|
||||
deps.submissionTimeout = 50 * time.Millisecond
|
||||
@@ -1744,7 +1745,7 @@ func TestServeTaskPluginImageProtocolWaitsForAsynchronousTask(t *testing.T) {
|
||||
return nil
|
||||
}
|
||||
task.Status = tc.terminal
|
||||
task.FailReason = "DataInspectionFailed"
|
||||
task.FailReason = "DataInspectionFailed: see https://private.example.com/jobs/1"
|
||||
task.SetData(map[string]any{"images": []string{"https://cdn.example/polled.png"}})
|
||||
return nil
|
||||
}
|
||||
@@ -1763,6 +1764,7 @@ func TestServeTaskPluginImageProtocolWaitsForAsynchronousTask(t *testing.T) {
|
||||
errorBody := response["error"].(map[string]any)
|
||||
assert.Equal(t, tc.wantCode, errorBody["code"])
|
||||
assert.Contains(t, errorBody["message"], tc.wantMessage)
|
||||
assert.NotContains(t, errorBody["message"], "private.example.com")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
+4
-3
@@ -92,17 +92,18 @@ func Relay(c *gin.Context, relayFormat types.RelayFormat) {
|
||||
service.RecordRequestPolicyTermination(c, newAPIError)
|
||||
logger.LogError(c, common.LogText("relay error: %s", common.LocalLogPreview(newAPIError.Error())))
|
||||
newAPIError.SetMessage(common.MessageWithRequestId(newAPIError.Error(), requestId))
|
||||
upstreamURLs := common.UpstreamAddresses(c)
|
||||
switch relayFormat {
|
||||
case types.RelayFormatOpenAIRealtime:
|
||||
helper.WssError(c, ws, newAPIError.ToOpenAIError())
|
||||
helper.WssError(c, ws, newAPIError.ToOpenAIError(upstreamURLs...))
|
||||
case types.RelayFormatClaude:
|
||||
c.JSON(newAPIError.StatusCode, gin.H{
|
||||
"type": "error",
|
||||
"error": newAPIError.ToClaudeError(),
|
||||
"error": newAPIError.ToClaudeError(upstreamURLs...),
|
||||
})
|
||||
default:
|
||||
c.JSON(newAPIError.StatusCode, gin.H{
|
||||
"error": newAPIError.ToOpenAIError(),
|
||||
"error": newAPIError.ToOpenAIError(upstreamURLs...),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -123,6 +123,6 @@ func ResponsesWebSocket(c *gin.Context) {
|
||||
if apiError := relay.ResponsesWebSocketHelper(c, ws, runner); apiError != nil {
|
||||
logger.LogError(c, common.LogText("responses websocket relay error: %s", common.LocalLogPreview(apiError.Error())))
|
||||
apiError.SetMessage(common.MessageWithRequestId(apiError.Error(), requestID))
|
||||
helper.WssError(c, ws, apiError.ToOpenAIError())
|
||||
helper.WssError(c, ws, apiError.ToOpenAIError(common.UpstreamAddresses(c)...))
|
||||
}
|
||||
}
|
||||
@@ -54,6 +54,47 @@ func TestAdaptorUsesExactRouteAndQueryAuth(t *testing.T) {
|
||||
assert.Equal(t, "sk-test", parsedURL.Query().Get("api_key"))
|
||||
}
|
||||
|
||||
// An absolute route sends the request to a host other than the channel base
|
||||
// URL, and a channel proxy is dialed before the upstream; a failed handshake
|
||||
// must hide both.
|
||||
func TestAdaptorMasksDialedHostsInDialFailure(t *testing.T) {
|
||||
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusForbidden)
|
||||
}))
|
||||
defer upstream.Close()
|
||||
upstreamURL, err := url.Parse(upstream.URL)
|
||||
require.NoError(t, err)
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
proxy string
|
||||
hidden string
|
||||
}{
|
||||
{name: "absolute route host", hidden: "localhost"},
|
||||
{name: "channel proxy host", proxy: "http://egress.private.invalid:8080", hidden: "egress.private.invalid"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
info := advancedCustomRelayInfo(&dto.AdvancedCustomConfig{
|
||||
Routes: []dto.AdvancedCustomRoute{{IncomingPath: "/v1/realtime", UpstreamPath: "http://localhost:" + upstreamURL.Port() + "/v1/realtime", Converter: relayconvert.ConverterNone}},
|
||||
})
|
||||
info.RelayMode = relayconstant.RelayModeRealtime
|
||||
info.RequestURLPath = "/v1/realtime"
|
||||
info.ChannelSetting.Proxy = tc.proxy
|
||||
c := advancedCustomGinContext("/v1/realtime")
|
||||
common.SetContextKey(c, constant.ContextKeyChannelBaseUrl, info.ChannelBaseUrl)
|
||||
common.SetContextKey(c, constant.ContextKeyChannelSetting, info.ChannelSetting)
|
||||
|
||||
_, err := (&Adaptor{}).DoRequest(c, info, nil)
|
||||
|
||||
var apiErr *types.NewAPIError
|
||||
require.ErrorAs(t, err, &apiErr)
|
||||
require.Contains(t, apiErr.Error(), tc.hidden)
|
||||
message := apiErr.ToOpenAIError(common.UpstreamAddresses(c)...).Message
|
||||
assert.Contains(t, message, "dial failed to ws://***:"+upstreamURL.Port()+"/***/***")
|
||||
assert.NotContains(t, message, tc.hidden)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdaptorJoinsUpstreamPathWithChannelBaseURL(t *testing.T) {
|
||||
adaptor := &Adaptor{}
|
||||
info := advancedCustomRelayInfo(&dto.AdvancedCustomConfig{
|
||||
|
||||
@@ -6,12 +6,14 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
common2 "github.com/QuantumNous/new-api/common"
|
||||
appconstant "github.com/QuantumNous/new-api/constant"
|
||||
"github.com/QuantumNous/new-api/logger"
|
||||
"github.com/QuantumNous/new-api/relay/common"
|
||||
"github.com/QuantumNous/new-api/relay/constant"
|
||||
@@ -382,6 +384,9 @@ func DoWssRequest(a Adaptor, c *gin.Context, info *common.RelayInfo, requestBody
|
||||
return nil, fmt.Errorf("get request url failed: %w", err)
|
||||
}
|
||||
fullRequestURL = toWebSocketURL(fullRequestURL)
|
||||
if targetURL, parseErr := url.Parse(fullRequestURL); parseErr == nil {
|
||||
common2.SetContextKey(c, appconstant.ContextKeyUpstreamRequestOrigin, targetURL.Scheme+"://"+targetURL.Host)
|
||||
}
|
||||
targetHeader := http.Header{}
|
||||
err = a.SetupRequestHeader(c, &targetHeader, info)
|
||||
if err != nil {
|
||||
@@ -560,6 +565,7 @@ func doRequest(c *gin.Context, req *http.Request, info *common.RelayInfo) (*http
|
||||
}
|
||||
}
|
||||
|
||||
common2.SetContextKey(c, appconstant.ContextKeyUpstreamRequestOrigin, req.URL.Scheme+"://"+req.URL.Host)
|
||||
resp, err := relayClient.Do(req)
|
||||
if err != nil {
|
||||
logger.LogError(c, common2.LogText("do request failed: %s", err.Error()))
|
||||
|
||||
@@ -453,11 +453,11 @@ func GeminiChatHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.R
|
||||
case types.RelayFormatClaude:
|
||||
c.JSON(newAPIError.StatusCode, gin.H{
|
||||
"type": "error",
|
||||
"error": newAPIError.ToClaudeError(),
|
||||
"error": newAPIError.ToClaudeError(common.UpstreamAddresses(c)...),
|
||||
})
|
||||
default:
|
||||
c.JSON(newAPIError.StatusCode, gin.H{
|
||||
"error": newAPIError.ToOpenAIError(),
|
||||
"error": newAPIError.ToOpenAIError(common.UpstreamAddresses(c)...),
|
||||
})
|
||||
}
|
||||
return &usage, nil
|
||||
|
||||
@@ -341,9 +341,9 @@ func OaiResponsesToChatStreamHandler(c *gin.Context, info *relaycommon.RelayInfo
|
||||
logger.LogError(c, common.LogText("responses stream failed after output was delivered, billing the delivered output: %s", streamErr.Error()))
|
||||
streamErr.SetMessage(common.MessageWithRequestId(streamErr.Error(), c.GetString(common.RequestIdKey)))
|
||||
if info.RelayFormat == types.RelayFormatClaude {
|
||||
c.JSON(streamErr.StatusCode, gin.H{"type": "error", "error": streamErr.ToClaudeError()})
|
||||
c.JSON(streamErr.StatusCode, gin.H{"type": "error", "error": streamErr.ToClaudeError(common.UpstreamAddresses(c)...)})
|
||||
} else {
|
||||
c.JSON(streamErr.StatusCode, gin.H{"error": streamErr.ToOpenAIError()})
|
||||
c.JSON(streamErr.StatusCode, gin.H{"error": streamErr.ToOpenAIError(common.UpstreamAddresses(c)...)})
|
||||
}
|
||||
return usage, nil
|
||||
}
|
||||
|
||||
@@ -165,7 +165,12 @@ func (a *TaskAdaptor) ValidateRequestAndSetAction(c *gin.Context, info *relaycom
|
||||
}
|
||||
}
|
||||
if _, err := a.buildSubmit(c, info); err != nil {
|
||||
return service.TaskErrorWrapperLocal(err, "plugin_request_invalid", http.StatusBadRequest)
|
||||
// utils.fetch failures in the build hook name the host fetched: the
|
||||
// channel base URL or one of the plugin's allowed hosts, which an
|
||||
// installed plugin may point at private hosts.
|
||||
taskErr := service.TaskErrorWrapper(err, "plugin_request_invalid", http.StatusBadRequest, append(common.UpstreamAddresses(c), a.plugin.Meta.AllowedHosts...)...)
|
||||
taskErr.LocalError = true
|
||||
return taskErr
|
||||
}
|
||||
// The descriptor may rewrite the model. Validate profiled requests against
|
||||
// that final model before any quota calculation or upstream submission; an
|
||||
@@ -535,7 +540,7 @@ func (a *TaskAdaptor) ParseResponse(c *gin.Context, resp *http.Response, info *r
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
failure := service.TaskErrorWrapper(err, "read_response_body_failed", http.StatusBadGateway)
|
||||
failure := service.TaskErrorWrapper(err, "read_response_body_failed", http.StatusBadGateway, common.UpstreamAddresses(c)...)
|
||||
failure.NoRetry = streaming
|
||||
return nil, failure
|
||||
}
|
||||
@@ -550,7 +555,7 @@ func (a *TaskAdaptor) ParseResponse(c *gin.Context, resp *http.Response, info *r
|
||||
resp.StatusCode,
|
||||
time.Since(started).Milliseconds(),
|
||||
)
|
||||
return nil, service.TaskErrorWrapper(err, "plugin_submit_response_failed", http.StatusBadGateway)
|
||||
return nil, service.TaskErrorWrapper(err, "plugin_submit_response_failed", http.StatusBadGateway, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
if object, ok := value.(map[string]any); ok {
|
||||
if _, forbidden := object["clientResponse"]; forbidden {
|
||||
@@ -569,7 +574,7 @@ func (a *TaskAdaptor) ParseResponse(c *gin.Context, resp *http.Response, info *r
|
||||
resp.StatusCode,
|
||||
time.Since(started).Milliseconds(),
|
||||
)
|
||||
return nil, service.TaskErrorWrapper(err, "plugin_submit_response_invalid", http.StatusBadGateway)
|
||||
return nil, service.TaskErrorWrapper(err, "plugin_submit_response_invalid", http.StatusBadGateway, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
var taskData []byte
|
||||
if parsed.TaskData != nil {
|
||||
@@ -578,7 +583,7 @@ func (a *TaskAdaptor) ParseResponse(c *gin.Context, resp *http.Response, info *r
|
||||
if err == nil {
|
||||
err = fmt.Errorf("task data exceeds size limit")
|
||||
}
|
||||
return nil, service.TaskErrorWrapper(err, "plugin_submit_response_invalid", http.StatusBadGateway)
|
||||
return nil, service.TaskErrorWrapper(err, "plugin_submit_response_invalid", http.StatusBadGateway, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
}
|
||||
pluginState, _ := encodeReturnedPluginState(value)
|
||||
|
||||
@@ -412,6 +412,32 @@ export function buildSubmitRequest(ctx){return {url:ctx.baseUrl+"/submit"}} expo
|
||||
assert.Contains(t, taskErr.Message, "must not return renderer")
|
||||
}
|
||||
|
||||
func TestTaskAdaptorMasksChannelAddressInBuildFailure(t *testing.T) {
|
||||
source := `
|
||||
export const meta = {apiVersion:1,key:"build-fail",name:"Build Fail",version:"1.0.0",author:{name:"Test"},models:["m"],fetchMode:"per_task",allowedHosts:["assets.corp.internal"]};
|
||||
export function buildSubmitRequest(ctx){throw new Error("utils.fetch to private.example.com failed: dial tcp: lookup private.example.com: no such host; utils.fetch to assets.corp.internal failed: dial tcp: lookup assets.corp.internal: no such host")}
|
||||
export function parseSubmitResponse(){return {taskId:"one"}} export function buildQueryRequest(){return {}} export function parseTaskResult(){return {status:"SUCCESS"}}
|
||||
`
|
||||
plugin, err := pluginruntime.NewRegistry().Register(source, pluginruntime.Options{})
|
||||
require.NoError(t, err)
|
||||
c, _ := gin.CreateTestContext(httptest.NewRecorder())
|
||||
c.Request = httptest.NewRequest(http.MethodPost, "/v1/video/generations", nil)
|
||||
c.Set("task_request", map[string]any{"model": "m"})
|
||||
common.SetContextKey(c, constant.ContextKeyChannelBaseUrl, "https://private.example.com")
|
||||
info := &relaycommon.RelayInfo{ChannelMeta: &relaycommon.ChannelMeta{ChannelBaseUrl: "https://private.example.com"}, TaskRelayInfo: &relaycommon.TaskRelayInfo{}, OriginModelName: "m"}
|
||||
adaptor := New(plugin)
|
||||
adaptor.Init(info)
|
||||
|
||||
taskErr := adaptor.ValidateRequestAndSetAction(c, info)
|
||||
|
||||
require.NotNil(t, taskErr)
|
||||
assert.Equal(t, "plugin_request_invalid", taskErr.Code)
|
||||
assert.True(t, taskErr.LocalError)
|
||||
assert.Contains(t, taskErr.Message, "no such host")
|
||||
assert.NotContains(t, taskErr.Message, "private.example.com")
|
||||
assert.NotContains(t, taskErr.Message, "assets.corp.internal")
|
||||
}
|
||||
|
||||
const orderedProtocolPlugin = `
|
||||
export const meta = {apiVersion:1,key:"ordered",name:"Ordered",version:"1.0.0",author:{name:"Test"},models:["m"],fetchMode:"per_task",requiredCapabilities:["json-order@1"],protocols:[{name:"openai_responses",supports:["sync"]}]};
|
||||
export const protocols = {openai_responses:{decodeRequest:function(ctx){const b = ctx.body.value; return {kind:"submit",model:ctx.model,requestBody:{model:b.model,state:b.state,questions:b.questions}};},renderFinal:function(){return {};}}};
|
||||
|
||||
+10
-10
@@ -66,7 +66,7 @@ func ResolveOriginTask(c *gin.Context, info *relaycommon.RelayInfo) *dto.TaskErr
|
||||
// 查找原始任务
|
||||
originTask, exist, err := model.GetByTaskId(info.UserId, info.OriginTaskID)
|
||||
if err != nil {
|
||||
return service.TaskErrorWrapper(err, "get_origin_task_failed", http.StatusInternalServerError)
|
||||
return service.TaskErrorWrapper(err, "get_origin_task_failed", http.StatusInternalServerError, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
if !exist {
|
||||
return service.TaskErrorWrapperLocal(errors.New("task_origin_not_exist"), "task_not_exist", http.StatusBadRequest)
|
||||
@@ -100,7 +100,7 @@ func ResolveOriginTask(c *gin.Context, info *relaycommon.RelayInfo) *dto.TaskErr
|
||||
if originTask.ChannelId != info.ChannelId {
|
||||
key, _, newAPIError := ch.GetNextEnabledKey()
|
||||
if newAPIError != nil {
|
||||
return service.TaskErrorWrapper(newAPIError, "channel_no_available_key", newAPIError.StatusCode)
|
||||
return service.TaskErrorWrapper(newAPIError, "channel_no_available_key", newAPIError.StatusCode, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
common.SetContextKey(c, constant.ContextKeyChannelKey, key)
|
||||
common.SetContextKey(c, constant.ContextKeyChannelType, ch.Type)
|
||||
@@ -261,16 +261,16 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitRe
|
||||
if useTiered {
|
||||
provider, supported := adaptor.(channel.TaskUsageFactsProvider)
|
||||
if billingexpr.UsesFixedPricing(exprStr) {
|
||||
return nil, service.TaskErrorWrapper(fmt.Errorf("fixed pricing is not supported for task usage expressions"), "model_price_error", http.StatusBadRequest)
|
||||
return nil, service.TaskErrorWrapper(fmt.Errorf("fixed pricing is not supported for task usage expressions"), "model_price_error", http.StatusBadRequest, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
if !exists || !supported {
|
||||
return nil, service.TaskErrorWrapper(fmt.Errorf("task model %s has no usage expression or meter", modelName), "model_price_error", http.StatusBadRequest)
|
||||
return nil, service.TaskErrorWrapper(fmt.Errorf("task model %s has no usage expression or meter", modelName), "model_price_error", http.StatusBadRequest, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
sharedModel := pinnedPlugin.Generation.SharedModel(modelName) || pinnedPlugin.Generation.SharedModel(info.UpstreamModelName)
|
||||
if sharedModel && pinnedPlugin.Plugin != nil {
|
||||
schema, _ := pinnedPlugin.Plugin.Meta.UsageForModels(info.UpstreamModelName, modelName)
|
||||
if !billing_setting.TaskExprCompatible(exprStr, schema) {
|
||||
return nil, service.TaskErrorWrapper(fmt.Errorf("task model %s pricing is not configured for plugin %s", modelName, pluginKey), "model_price_error", http.StatusBadRequest)
|
||||
return nil, service.TaskErrorWrapper(fmt.Errorf("task model %s pricing is not configured for plugin %s", modelName, pluginKey), "model_price_error", http.StatusBadRequest, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
}
|
||||
var facts map[string]any
|
||||
@@ -287,7 +287,7 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitRe
|
||||
if runErr == nil {
|
||||
runErr = fmt.Errorf("negative task expression result")
|
||||
}
|
||||
return nil, service.TaskErrorWrapper(runErr, "model_price_error", http.StatusBadRequest)
|
||||
return nil, service.TaskErrorWrapper(runErr, "model_price_error", http.StatusBadRequest, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
groupRatioInfo := helper.HandleGroupRatio(c, info)
|
||||
quota, clamp := common.QuotaRoundChecked(cost * common.QuotaPerUnit * groupRatioInfo.GroupRatio)
|
||||
@@ -297,7 +297,7 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitRe
|
||||
} else {
|
||||
priceData, err = helper.ModelPriceHelperPerCall(c, info)
|
||||
if err != nil {
|
||||
return nil, service.TaskErrorWrapper(err, "model_price_error", http.StatusBadRequest)
|
||||
return nil, service.TaskErrorWrapper(err, "model_price_error", http.StatusBadRequest, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
}
|
||||
info.PriceData = priceData
|
||||
@@ -341,13 +341,13 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitRe
|
||||
// 8. 构建请求体
|
||||
requestBody, err := adaptor.BuildRequestBody(c, info)
|
||||
if err != nil {
|
||||
return nil, service.TaskErrorWrapper(err, "build_request_failed", http.StatusInternalServerError)
|
||||
return nil, service.TaskErrorWrapper(err, "build_request_failed", http.StatusInternalServerError, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
|
||||
// 9. 发送请求
|
||||
resp, err := adaptor.DoRequest(c, info, requestBody)
|
||||
if err != nil {
|
||||
return nil, service.TaskErrorWrapper(err, "do_request_failed", http.StatusInternalServerError)
|
||||
return nil, service.TaskErrorWrapper(err, "do_request_failed", http.StatusInternalServerError, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
if resp == nil {
|
||||
return nil, service.TaskErrorWrapperLocal(errors.New("upstream returned an empty response"), "fail_to_fetch_task", http.StatusBadGateway)
|
||||
@@ -357,7 +357,7 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitRe
|
||||
// or 202 Accepted, and parseSubmitResponse receives the exact status code.
|
||||
if resp.StatusCode/100 != 2 {
|
||||
responseBody, _ := io.ReadAll(resp.Body)
|
||||
return nil, service.TaskErrorWrapper(fmt.Errorf("%s", string(responseBody)), "fail_to_fetch_task", resp.StatusCode)
|
||||
return nil, service.TaskErrorWrapper(fmt.Errorf("%s", string(responseBody)), "fail_to_fetch_task", resp.StatusCode, common.UpstreamAddresses(c)...)
|
||||
}
|
||||
|
||||
// 10. Parse only. The controller presents the response after the durable
|
||||
|
||||
@@ -181,6 +181,7 @@ func ResponsesWebSocketHelper(c *gin.Context, client *websocket.Conn, runner Res
|
||||
func (s *responsesWSSession) runRequest(state *responsesWSCallState, message []byte, envelope responsesWSCreateEvent, streamID string, requestID string) {
|
||||
create, parseErr := normalizeResponsesWSCreateEvent(message, envelope, streamID)
|
||||
var apiErr *types.NewAPIError
|
||||
var upstreamURLs []string
|
||||
defer func() {
|
||||
if recovered := recover(); recovered != nil {
|
||||
apiErr = types.NewError(fmt.Errorf("responses websocket request panic: %v", recovered), types.ErrorCodeBadResponse, types.ErrOptionWithSkipRetry())
|
||||
@@ -193,7 +194,7 @@ func (s *responsesWSSession) runRequest(state *responsesWSCallState, message []b
|
||||
// as a normal closure; shutdown itself takes neither lock.
|
||||
outgoing := state.terminal
|
||||
if apiErr != nil {
|
||||
if body, err := buildResponsesWSErrorPayload(envelope.EventID, streamID, apiErr); err == nil {
|
||||
if body, err := buildResponsesWSErrorPayload(envelope.EventID, streamID, apiErr, upstreamURLs...); err == nil {
|
||||
outgoing = &responsesWSMessage{kind: websocket.TextMessage, body: body}
|
||||
}
|
||||
}
|
||||
@@ -224,7 +225,9 @@ func (s *responsesWSSession) runRequest(state *responsesWSCallState, message []b
|
||||
}
|
||||
c.Request.Body = io.NopCloser(bytes.NewReader(create.Body))
|
||||
c.Request.ContentLength = int64(len(create.Body))
|
||||
return s.runCall(c, state, create)
|
||||
callErr := s.runCall(c, state, create)
|
||||
upstreamURLs = common.UpstreamAddresses(c)
|
||||
return callErr
|
||||
})
|
||||
if apiErr != nil && (apiErr.StatusCode == http.StatusUnauthorized || apiErr.StatusCode == http.StatusForbidden) {
|
||||
state.closeCode = websocket.ClosePolicyViolation
|
||||
@@ -348,6 +351,7 @@ func (s *responsesWSSession) runCall(c *gin.Context, state *responsesWSCallState
|
||||
appconstant.ContextKeyChannelParamOverride, appconstant.ContextKeyChannelHeaderOverride, appconstant.ContextKeyChannelOrganization,
|
||||
appconstant.ContextKeyChannelAutoBan, appconstant.ContextKeyChannelModelMapping, appconstant.ContextKeyChannelStatusCodeMapping,
|
||||
appconstant.ContextKeyChannelIsMultiKey, appconstant.ContextKeyChannelMultiKeyIndex, appconstant.ContextKeyChannelKey, appconstant.ContextKeyChannelBaseUrl,
|
||||
appconstant.ContextKeyUpstreamRequestOrigin,
|
||||
} {
|
||||
if value, ok := c.Get(string(key)); ok {
|
||||
s.lockedContext[key] = value
|
||||
@@ -880,7 +884,7 @@ func buildResponsesWSCreateEvent(jsonData []byte, generate common.RawMessage, st
|
||||
return common.Marshal(event)
|
||||
}
|
||||
|
||||
func buildResponsesWSErrorPayload(eventID, streamID string, apiErr *types.NewAPIError) ([]byte, error) {
|
||||
func buildResponsesWSErrorPayload(eventID, streamID string, apiErr *types.NewAPIError, upstreamURLs ...string) ([]byte, error) {
|
||||
if apiErr == nil {
|
||||
return nil, errors.New("api error is nil")
|
||||
}
|
||||
@@ -888,7 +892,7 @@ func buildResponsesWSErrorPayload(eventID, streamID string, apiErr *types.NewAPI
|
||||
if status == 0 {
|
||||
status = http.StatusInternalServerError
|
||||
}
|
||||
openaiErr := apiErr.ToOpenAIError()
|
||||
openaiErr := apiErr.ToOpenAIError(upstreamURLs...)
|
||||
return common.Marshal(&responsesWSErrorEvent{
|
||||
Type: "error",
|
||||
Status: status,
|
||||
|
||||
@@ -1,15 +1,25 @@
|
||||
package kitutil
|
||||
|
||||
import (
|
||||
"maps"
|
||||
"net"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
)
|
||||
|
||||
var (
|
||||
maskURLPattern = regexp.MustCompile(`(http|https)://[^\s/$.?#].[^\s]*`)
|
||||
maskDomainPattern = regexp.MustCompile(`\b(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}\b`)
|
||||
maskIPPattern = regexp.MustCompile(`\b(?:\d{1,3}\.){3}\d{1,3}\b`)
|
||||
// maskURLPattern stops at characters a URL cannot contain unescaped, such
|
||||
// as double quotes and angle or curly brackets, so a URL quoted in an error
|
||||
// keeps its delimiters. Apostrophes and parentheses are valid inside a URL;
|
||||
// the caller trims them, like other punctuation, only when they trail it.
|
||||
maskURLPattern = regexp.MustCompile(`(?:https?|wss?)://[^\s"<>{}]+`)
|
||||
maskIPPattern = regexp.MustCompile(`\b(?:\d{1,3}\.){3}\d{1,3}\b`)
|
||||
// maskBracketedIPv6Pattern matches the bracketed form Go network errors use
|
||||
// for IPv6 addresses, as in "dial tcp [2001:db8::1]:443", with an optional
|
||||
// zone; the caller confirms the content parses as an IP.
|
||||
maskBracketedIPv6Pattern = regexp.MustCompile(`\[([0-9A-Fa-f:.]*:[0-9A-Fa-f:.]*)(%[^\]\s]*)?\]`)
|
||||
// maskApiKeyPattern matches patterns like 'api_key:xxx' or "api_key:xxx" to mask the API key value
|
||||
maskApiKeyPattern = regexp.MustCompile(`(['"]?)api_key:([^\s'"]+)(['"]?)`)
|
||||
)
|
||||
@@ -53,30 +63,111 @@ func maskHostForPlainDomain(domain string) string {
|
||||
return stars + "." + strings.Join(tail, ".")
|
||||
}
|
||||
|
||||
// MaskSensitiveInfo masks sensitive information like URLs, IPs, and domain names in a string
|
||||
// Example:
|
||||
// http://example.com -> http://***.com
|
||||
// https://api.test.org/v1/users/123?key=secret -> https://***.org/***/***/?key=***
|
||||
// https://sub.domain.co.uk/path/to/resource -> https://***.co.uk/***/***
|
||||
// 192.168.1.1 -> ***.***.***.***
|
||||
// openai.com -> ***.com
|
||||
// www.openai.com -> ***.***.com
|
||||
// api.openai.com -> ***.***.com
|
||||
func MaskSensitiveInfo(str string) string {
|
||||
// Mask URLs
|
||||
str = maskURLPattern.ReplaceAllStringFunc(str, func(urlStr string) string {
|
||||
// maskURLQuery keeps the parameter names of a raw query and masks the values.
|
||||
// Example: key=secret&alt=sse -> ?alt=***&key=***
|
||||
func maskURLQuery(rawQuery string) string {
|
||||
if rawQuery == "" {
|
||||
return ""
|
||||
}
|
||||
values, err := url.ParseQuery(rawQuery)
|
||||
if err != nil {
|
||||
// If can't parse query, just mask the whole query string
|
||||
return "?***"
|
||||
}
|
||||
if len(values) == 0 {
|
||||
return ""
|
||||
}
|
||||
maskedParams := make([]string, 0, len(values))
|
||||
for _, key := range slices.Sorted(maps.Keys(values)) {
|
||||
maskedParams = append(maskedParams, key+"=***")
|
||||
}
|
||||
return "?" + strings.Join(maskedParams, "&")
|
||||
}
|
||||
|
||||
// MaskSensitiveInfo masks the upstream addresses in str and leaves every other
|
||||
// URL and domain readable. upstreamURLs are the channel base URL and the URL
|
||||
// the gateway actually requested, which differs for routes such as absolute
|
||||
// Advanced Custom paths; empty entries are skipped. Credentials and IPv4
|
||||
// addresses are masked regardless of the upstream: API key values, the query
|
||||
// values and userinfo of every URL, which carry secrets such as client_secret
|
||||
// or key on provider endpoints, and IPs, which in network errors include the
|
||||
// gateway's own address.
|
||||
// Example with upstream https://api.upstream.example.com:
|
||||
// https://api.upstream.example.com/v1/chat?key=secret -> https://***.com/***/***?key=***
|
||||
// lookup api.upstream.example.com -> lookup ***.***.***.com
|
||||
// dial tcp 192.168.1.1:443 -> dial tcp ***.***.***.***:443
|
||||
// https://docs.example.org/errors -> unchanged
|
||||
// https://auth.example.org/token?client_secret=s -> https://auth.example.org/token?client_secret=***
|
||||
func MaskSensitiveInfo(str string, upstreamURLs ...string) string {
|
||||
// Mask API keys (e.g., "api_key:AIzaSyAAAaUooTUni8AdaOkSRMda30n_Q4vrV70" -> "api_key:***")
|
||||
str = maskApiKeyPattern.ReplaceAllString(str, "${1}api_key:***${3}")
|
||||
|
||||
upstreamHosts := make([]string, 0, len(upstreamURLs))
|
||||
for _, upstreamURL := range upstreamURLs {
|
||||
upstreamURL = strings.TrimSpace(upstreamURL)
|
||||
if upstreamURL == "" {
|
||||
continue
|
||||
}
|
||||
if !strings.Contains(upstreamURL, "://") {
|
||||
// A base URL saved without a scheme would otherwise parse as a path.
|
||||
upstreamURL = "//" + upstreamURL
|
||||
}
|
||||
parsed, err := url.Parse(upstreamURL)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
// A fully qualified host keeps its trailing root dot; errors may omit it.
|
||||
if host := strings.TrimSuffix(parsed.Hostname(), "."); host != "" {
|
||||
upstreamHosts = append(upstreamHosts, host)
|
||||
}
|
||||
}
|
||||
// Mask URLs on an upstream host, and the query values of every other URL
|
||||
str = maskURLPattern.ReplaceAllStringFunc(str, func(match string) string {
|
||||
urlStr := match
|
||||
for urlStr != "" {
|
||||
last := urlStr[len(urlStr)-1]
|
||||
if strings.IndexByte(".,;:!?'", last) < 0 && (last != ')' || strings.Count(urlStr, "(") >= strings.Count(urlStr, ")")) {
|
||||
break
|
||||
}
|
||||
urlStr = urlStr[:len(urlStr)-1]
|
||||
}
|
||||
suffix := match[len(urlStr):]
|
||||
u, err := url.Parse(urlStr)
|
||||
if err != nil {
|
||||
return urlStr
|
||||
return match
|
||||
}
|
||||
host := strings.TrimSuffix(u.Hostname(), ".")
|
||||
if !slices.ContainsFunc(upstreamHosts, func(upstreamHost string) bool { return strings.EqualFold(upstreamHost, host) }) {
|
||||
if u.RawQuery == "" && u.User == nil {
|
||||
return match
|
||||
}
|
||||
// Userinfo is a credential wherever the URL points.
|
||||
if u.User != nil {
|
||||
scheme, afterScheme, _ := strings.Cut(urlStr, "://")
|
||||
authorityEnd := strings.IndexAny(afterScheme, "/?#")
|
||||
if authorityEnd < 0 {
|
||||
authorityEnd = len(afterScheme)
|
||||
}
|
||||
urlStr = scheme + "://***@" + afterScheme[strings.LastIndex(afterScheme[:authorityEnd], "@")+1:]
|
||||
}
|
||||
if u.RawQuery == "" {
|
||||
return urlStr + suffix
|
||||
}
|
||||
beforeQuery, _, _ := strings.Cut(urlStr, "?")
|
||||
fragment := ""
|
||||
if u.Fragment != "" {
|
||||
fragment = "#" + u.EscapedFragment()
|
||||
}
|
||||
return beforeQuery + maskURLQuery(u.RawQuery) + fragment + suffix
|
||||
}
|
||||
|
||||
host := u.Host
|
||||
if host == "" {
|
||||
return urlStr
|
||||
}
|
||||
|
||||
// Mask host with unified logic
|
||||
maskedHost := maskHostForURL(host)
|
||||
if ip := net.ParseIP(host); ip != nil && ip.To4() != nil {
|
||||
maskedHost = "***.***.***.***"
|
||||
}
|
||||
if port := u.Port(); port != "" {
|
||||
maskedHost += ":" + port
|
||||
}
|
||||
|
||||
result := u.Scheme + "://" + maskedHost
|
||||
|
||||
@@ -96,36 +187,35 @@ func MaskSensitiveInfo(str string) string {
|
||||
result += "/"
|
||||
}
|
||||
|
||||
// Mask query parameters
|
||||
if u.RawQuery != "" {
|
||||
values, err := url.ParseQuery(u.RawQuery)
|
||||
if err != nil {
|
||||
// If can't parse query, just mask the whole query string
|
||||
result += "?***"
|
||||
} else {
|
||||
maskedParams := make([]string, 0, len(values))
|
||||
for key := range values {
|
||||
maskedParams = append(maskedParams, key+"=***")
|
||||
}
|
||||
if len(maskedParams) > 0 {
|
||||
result += "?" + strings.Join(maskedParams, "&")
|
||||
}
|
||||
}
|
||||
return result + maskURLQuery(u.RawQuery) + suffix
|
||||
})
|
||||
|
||||
// Mask IP addresses: in network errors they are the resolved upstream
|
||||
// address or the gateway's own, so they are masked even without an
|
||||
// upstream address, and an IP upstream host is covered too. IPv6 goes
|
||||
// first so an embedded IPv4 such as [::ffff:10.0.0.1] still parses.
|
||||
str = maskBracketedIPv6Pattern.ReplaceAllStringFunc(str, func(match string) string {
|
||||
if net.ParseIP(maskBracketedIPv6Pattern.FindStringSubmatch(match)[1]) == nil {
|
||||
return match
|
||||
}
|
||||
|
||||
return result
|
||||
return "[***]"
|
||||
})
|
||||
|
||||
// Mask domain names without protocol (like openai.com, www.openai.com)
|
||||
str = maskDomainPattern.ReplaceAllStringFunc(str, func(domain string) string {
|
||||
return maskHostForPlainDomain(domain)
|
||||
})
|
||||
|
||||
// Mask IP addresses
|
||||
str = maskIPPattern.ReplaceAllString(str, "***.***.***.***")
|
||||
if len(upstreamHosts) == 0 {
|
||||
return str
|
||||
}
|
||||
|
||||
// Mask API keys (e.g., "api_key:AIzaSyAAAaUooTUni8AdaOkSRMda30n_Q4vrV70" -> "api_key:***")
|
||||
str = maskApiKeyPattern.ReplaceAllString(str, "${1}api_key:***${3}")
|
||||
// Mask an upstream host where it appears without a scheme, as in
|
||||
// "dial tcp: lookup api.example.com: no such host". Longer hosts go first
|
||||
// so a subdomain upstream is masked whole before its parent base host.
|
||||
slices.SortFunc(upstreamHosts, func(a, b string) int { return len(b) - len(a) })
|
||||
for _, host := range upstreamHosts {
|
||||
if net.ParseIP(host) != nil {
|
||||
continue
|
||||
}
|
||||
hostPattern := regexp.MustCompile(`(?i)\b` + regexp.QuoteMeta(host) + `\b`)
|
||||
str = hostPattern.ReplaceAllStringFunc(str, maskHostForPlainDomain)
|
||||
}
|
||||
|
||||
return str
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
package kitutil
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestMaskSensitiveInfoMasksOnlyTheUpstreamAddress(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
upstreamURLs []string
|
||||
input string
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "no upstream address leaves hosts readable but masks IPs",
|
||||
upstreamURLs: []string{"", ""},
|
||||
input: `Post "https://api.upstream.example.com/v1/chat/completions": dial tcp 10.0.0.1:443: connect: connection refused`,
|
||||
want: `Post "https://api.upstream.example.com/v1/chat/completions": dial tcp ***.***.***.***:443: connect: connection refused`,
|
||||
},
|
||||
{
|
||||
name: "bracketed IPv6 addresses are masked without an upstream address",
|
||||
upstreamURLs: nil,
|
||||
input: `dial tcp [2a01:4f8::1]:443: i/o timeout; lookup api.example.com on [::1]:53; dial tcp [fe80::1%eth0]:443; dial tcp [::ffff:10.0.0.1]:443; items[0] and [1:2] stay`,
|
||||
want: `dial tcp [***]:443: i/o timeout; lookup api.example.com on [***]:53; dial tcp [***]:443; dial tcp [***]:443; items[0] and [1:2] stay`,
|
||||
},
|
||||
{
|
||||
name: "IPv6 upstream base URL in a WebSocket dial error",
|
||||
upstreamURLs: []string{"http://[2001:db8::1]:8080"},
|
||||
input: `dial failed to ws://[2001:db8::1]:8080/v1/realtime: dial tcp [2001:db8::1]:8080: connect: connection refused`,
|
||||
want: `dial failed to ws://***:8080/***/***: dial tcp [***]:8080: connect: connection refused`,
|
||||
},
|
||||
{
|
||||
name: "no upstream address still masks API key values",
|
||||
upstreamURLs: nil,
|
||||
input: `invalid api_key:AIzaSyAAAaUooTUni8AdaOkSRMda30n_Q4vrV70`,
|
||||
want: `invalid api_key:***`,
|
||||
},
|
||||
{
|
||||
name: "query values carry credentials on any host",
|
||||
upstreamURLs: []string{"https://custom.base.example"},
|
||||
input: `Post "https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=id-1&client_secret=secret-1": dial tcp: i/o timeout, see https://docs.example.org/errors?lang=en#retry`,
|
||||
want: `Post "https://aip.baidubce.com/oauth/2.0/token?client_id=***&client_secret=***&grant_type=***": dial tcp: i/o timeout, see https://docs.example.org/errors?lang=***#retry`,
|
||||
},
|
||||
{
|
||||
name: "parentheses inside a URL keep its query masked",
|
||||
upstreamURLs: nil,
|
||||
input: `failed to download https://cdn.example/image(1).png?token=secret (status 403)`,
|
||||
want: `failed to download https://cdn.example/image(1).png?token=*** (status 403)`,
|
||||
},
|
||||
{
|
||||
name: "apostrophes inside a URL keep its query and userinfo masked",
|
||||
upstreamURLs: nil,
|
||||
input: `failed to download https://cdn.example/John's-photo.png?token=secret from 'https://o'neil:pw@cdn.example/a?sig=1'`,
|
||||
want: `failed to download https://cdn.example/John's-photo.png?token=*** from 'https://***@cdn.example/a?sig=***'`,
|
||||
},
|
||||
{
|
||||
name: "userinfo is masked on any host",
|
||||
upstreamURLs: nil,
|
||||
input: `downloading https://alice:secret@cdn.example/file.png and https://bob@cdn.example/a?sig=1`,
|
||||
want: `downloading https://***@cdn.example/file.png and https://***@cdn.example/a?sig=***`,
|
||||
},
|
||||
{
|
||||
name: "query values are masked without an upstream address",
|
||||
upstreamURLs: nil,
|
||||
input: `request to https://generativelanguage.googleapis.com/v1beta/models?key=AIza-secret failed`,
|
||||
want: `request to https://generativelanguage.googleapis.com/v1beta/models?key=*** failed`,
|
||||
},
|
||||
{
|
||||
name: "upstream URL, its IP and other URLs",
|
||||
upstreamURLs: []string{"https://api.upstream.example.com/openai"},
|
||||
input: `request to https://api.upstream.example.com/openai/v1/chat/completions?key=secret failed: dial tcp 10.0.0.1:443: i/o timeout, see https://docs.example.org/errors`,
|
||||
want: `request to https://***.com/***/***/***/***?key=*** failed: dial tcp ***.***.***.***:443: i/o timeout, see https://docs.example.org/errors`,
|
||||
},
|
||||
{
|
||||
name: "bare upstream host in a DNS error",
|
||||
upstreamURLs: []string{"https://API.Upstream.example.com"},
|
||||
input: `dial tcp: lookup api.upstream.example.com: no such host`,
|
||||
want: `dial tcp: lookup ***.***.***.com: no such host`,
|
||||
},
|
||||
{
|
||||
name: "fully qualified base URL with a trailing dot",
|
||||
upstreamURLs: []string{"https://private.example.com./v1"},
|
||||
input: `request to https://private.example.com./v1/responses failed: dial tcp: lookup private.example.com.: no such host`,
|
||||
want: `request to https://***.com/***/*** failed: dial tcp: lookup ***.***.com.: no such host`,
|
||||
},
|
||||
{
|
||||
name: "base URL saved without a scheme",
|
||||
upstreamURLs: []string{"relay.example.co.uk"},
|
||||
input: `request to https://relay.example.co.uk/v1/messages failed: EOF`,
|
||||
want: `request to https://***.co.uk/***/*** failed: EOF`,
|
||||
},
|
||||
{
|
||||
name: "quoted URL keeps its delimiters and trailing punctuation",
|
||||
upstreamURLs: []string{"https://api.upstream.example.com"},
|
||||
input: `Post "https://api.upstream.example.com/v1/chat/completions": EOF; retry https://api.upstream.example.com/v1/models.`,
|
||||
want: `Post "https://***.com/***/***/***": EOF; retry https://***.com/***/***.`,
|
||||
},
|
||||
{
|
||||
name: "port does not hide a country-code TLD",
|
||||
upstreamURLs: []string{"https://relay.example.co.uk:8443"},
|
||||
input: `request to https://relay.example.co.uk:8443/v1/messages failed`,
|
||||
want: `request to https://***.co.uk:8443/***/*** failed`,
|
||||
},
|
||||
{
|
||||
name: "IP upstream host is fully masked",
|
||||
upstreamURLs: []string{"http://10.0.0.1:8080"},
|
||||
input: `request to http://10.0.0.1:8080/v1/chat failed`,
|
||||
want: `request to http://***.***.***.***:8080/***/*** failed`,
|
||||
},
|
||||
{
|
||||
name: "requested origin of an absolute route besides the base URL",
|
||||
upstreamURLs: []string{"https://api.upstream.example.com", "wss://realtime.private.example.net"},
|
||||
input: `dial failed to wss://realtime.private.example.net/v1/responses: dial tcp: lookup realtime.private.example.net: no such host`,
|
||||
want: `dial failed to wss://***.net/***/***: dial tcp: lookup ***.***.***.net: no such host`,
|
||||
},
|
||||
{
|
||||
name: "subdomain upstream is masked whole before its parent base host",
|
||||
upstreamURLs: []string{"https://api.example.com", "wss://tenant.api.example.com"},
|
||||
input: `dial tcp: lookup tenant.api.example.com: no such host`,
|
||||
want: `dial tcp: lookup ***.***.***.com: no such host`,
|
||||
},
|
||||
{
|
||||
name: "other domains and dotted identifiers stay readable",
|
||||
upstreamURLs: []string{"https://api.upstream.example.com"},
|
||||
input: `Unknown event response.output_text.delta from model gpt-4.1, contact support@example.org`,
|
||||
want: `Unknown event response.output_text.delta from model gpt-4.1, contact support@example.org`,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
assert.Equal(t, tt.want, MaskSensitiveInfo(tt.input, tt.upstreamURLs...))
|
||||
})
|
||||
}
|
||||
}
|
||||
+13
-8
@@ -145,7 +145,9 @@ func (e *NewAPIError) ErrorWithStatusCode() string {
|
||||
return fmt.Sprintf("status_code=%d, %s", e.StatusCode, msg)
|
||||
}
|
||||
|
||||
func (e *NewAPIError) MaskSensitiveError() string {
|
||||
// MaskSensitiveError masks the upstream addresses of the channel that produced
|
||||
// the error in the error text; see kitutil.MaskSensitiveInfo.
|
||||
func (e *NewAPIError) MaskSensitiveError(upstreamURLs ...string) string {
|
||||
if e == nil {
|
||||
return ""
|
||||
}
|
||||
@@ -156,14 +158,14 @@ func (e *NewAPIError) MaskSensitiveError() string {
|
||||
if e.errorCode == ErrorCodeCountTokenFailed {
|
||||
return errStr
|
||||
}
|
||||
return kitutil.MaskSensitiveInfo(errStr)
|
||||
return kitutil.MaskSensitiveInfo(errStr, upstreamURLs...)
|
||||
}
|
||||
|
||||
func (e *NewAPIError) MaskSensitiveErrorWithStatusCode() string {
|
||||
func (e *NewAPIError) MaskSensitiveErrorWithStatusCode(upstreamURLs ...string) string {
|
||||
if e == nil {
|
||||
return ""
|
||||
}
|
||||
msg := e.MaskSensitiveError()
|
||||
msg := e.MaskSensitiveError(upstreamURLs...)
|
||||
if e.StatusCode == 0 {
|
||||
return msg
|
||||
}
|
||||
@@ -177,7 +179,9 @@ func (e *NewAPIError) SetMessage(message string) {
|
||||
e.Err = errors.New(message)
|
||||
}
|
||||
|
||||
func (e *NewAPIError) ToOpenAIError() OpenAIError {
|
||||
// ToOpenAIError masks the upstream addresses in the message; pass those of the
|
||||
// channel that produced the error, or none when no channel did.
|
||||
func (e *NewAPIError) ToOpenAIError(upstreamURLs ...string) OpenAIError {
|
||||
var result OpenAIError
|
||||
switch e.errorType {
|
||||
case ErrorTypeOpenAIError:
|
||||
@@ -202,7 +206,7 @@ func (e *NewAPIError) ToOpenAIError() OpenAIError {
|
||||
}
|
||||
}
|
||||
if e.errorCode != ErrorCodeCountTokenFailed {
|
||||
result.Message = kitutil.MaskSensitiveInfo(result.Message)
|
||||
result.Message = kitutil.MaskSensitiveInfo(result.Message, upstreamURLs...)
|
||||
}
|
||||
if result.Message == "" {
|
||||
result.Message = string(e.errorType)
|
||||
@@ -210,7 +214,8 @@ func (e *NewAPIError) ToOpenAIError() OpenAIError {
|
||||
return result
|
||||
}
|
||||
|
||||
func (e *NewAPIError) ToClaudeError() ClaudeError {
|
||||
// ToClaudeError masks the upstream addresses like ToOpenAIError.
|
||||
func (e *NewAPIError) ToClaudeError(upstreamURLs ...string) ClaudeError {
|
||||
var result ClaudeError
|
||||
switch e.errorType {
|
||||
case ErrorTypeOpenAIError:
|
||||
@@ -231,7 +236,7 @@ func (e *NewAPIError) ToClaudeError() ClaudeError {
|
||||
}
|
||||
}
|
||||
if e.errorCode != ErrorCodeCountTokenFailed {
|
||||
result.Message = kitutil.MaskSensitiveInfo(result.Message)
|
||||
result.Message = kitutil.MaskSensitiveInfo(result.Message, upstreamURLs...)
|
||||
}
|
||||
if result.Message == "" {
|
||||
result.Message = string(e.errorType)
|
||||
|
||||
+4
-2
@@ -197,13 +197,15 @@ func TaskErrorWrapperLocal(err error, code string, statusCode int) *taskdto.Task
|
||||
return openaiErr
|
||||
}
|
||||
|
||||
func TaskErrorWrapper(err error, code string, statusCode int) *taskdto.TaskError {
|
||||
// TaskErrorWrapper masks the upstream addresses of the task channel in network
|
||||
// errors; see common.MaskSensitiveInfo.
|
||||
func TaskErrorWrapper(err error, code string, statusCode int, upstreamURLs ...string) *taskdto.TaskError {
|
||||
text := err.Error()
|
||||
lowerText := strings.ToLower(text)
|
||||
if strings.Contains(lowerText, "post") || strings.Contains(lowerText, "dial") || strings.Contains(lowerText, "http") {
|
||||
common.SysLog(common.LogText("error: %s", text))
|
||||
//text = "请求上游地址失败"
|
||||
text = common.MaskSensitiveInfo(text)
|
||||
text = common.MaskSensitiveInfo(text, upstreamURLs...)
|
||||
}
|
||||
//避免暴露内部错误
|
||||
taskError := &taskdto.TaskError{
|
||||
|
||||
@@ -64,9 +64,10 @@ func ProcessChannelError(c *gin.Context, channelError types.ChannelError, err *t
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
logger.LogError(c, common.LogText("channel error (channel #%d, status code: %d): %s", channelError.ChannelId, err.StatusCode, common.LocalLogPreview(err.MaskSensitiveErrorWithStatusCode())))
|
||||
upstreamURLs := common.UpstreamAddresses(c)
|
||||
logger.LogError(c, common.LogText("channel error (channel #%d, status code: %d): %s", channelError.ChannelId, err.StatusCode, common.LocalLogPreview(err.MaskSensitiveErrorWithStatusCode(upstreamURLs...))))
|
||||
if ShouldDisableChannel(err) && channelError.AutoBan {
|
||||
reason := err.MaskSensitiveErrorWithStatusCode()
|
||||
reason := err.MaskSensitiveErrorWithStatusCode(upstreamURLs...)
|
||||
gopool.Go(func() {
|
||||
DisableChannel(channelError, reason)
|
||||
})
|
||||
@@ -93,6 +94,6 @@ func ProcessChannelError(c *gin.Context, channelError types.ChannelError, err *t
|
||||
startTime = time.Now()
|
||||
}
|
||||
useTimeSeconds := int(time.Since(startTime).Seconds())
|
||||
model.RecordErrorLog(c, userId, channelError.ChannelId, modelName, tokenName, err.MaskSensitiveErrorWithStatusCode(), tokenId, useTimeSeconds, common.GetContextKeyBool(c, constant.ContextKeyIsStream), userGroup, other)
|
||||
model.RecordErrorLog(c, userId, channelError.ChannelId, modelName, tokenName, err.MaskSensitiveErrorWithStatusCode(upstreamURLs...), tokenId, useTimeSeconds, common.GetContextKeyBool(c, constant.ContextKeyIsStream), userGroup, other)
|
||||
}
|
||||
}
|
||||
@@ -105,6 +105,7 @@ func TestProcessChannelErrorMasksDisableReasonAndNotification(t *testing.T) {
|
||||
notifyLimitStore.Delete(notifyKey)
|
||||
t.Cleanup(func() { notifyLimitStore.Delete(notifyKey) })
|
||||
c, _ := gin.CreateTestContext(httptest.NewRecorder())
|
||||
common.SetContextKey(c, constant.ContextKeyChannelBaseUrl, "https://private.example.com")
|
||||
apiErr := types.NewErrorWithStatusCode(errors.New("upstream https://private.example.com/path?token=review-token api_key:review-secret"), types.ErrorCodeChannelNoAvailableKey, http.StatusUnauthorized)
|
||||
ProcessChannelError(c, types.ChannelError{ChannelId: channel.Id, ChannelName: channel.Name, AutoBan: true}, apiErr, nil)
|
||||
var notification WebhookPayload
|
||||
@@ -125,6 +126,72 @@ func TestProcessChannelErrorMasksDisableReasonAndNotification(t *testing.T) {
|
||||
assert.Equal(t, http.StatusUnauthorized, apiErr.StatusCode)
|
||||
}
|
||||
|
||||
// Only addresses the admin configured are masked: a channel on the base URL
|
||||
// new-api presets reaches a public official endpoint, and so do adaptors that
|
||||
// build official endpoints themselves.
|
||||
func TestUpstreamAddressesMaskOnlyConfiguredAddresses(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
channelType int
|
||||
baseURL string
|
||||
origin string
|
||||
proxyEnv string
|
||||
message string
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "preset base URL stays readable",
|
||||
channelType: constant.ChannelTypeAnthropic,
|
||||
baseURL: "https://api.anthropic.com",
|
||||
origin: "https://api.anthropic.com",
|
||||
message: "request to https://api.anthropic.com/v1/messages failed, see https://platform.claude.com/docs/en/build-with-claude/task-budgets",
|
||||
want: "request to https://api.anthropic.com/v1/messages failed, see https://platform.claude.com/docs/en/build-with-claude/task-budgets",
|
||||
},
|
||||
{
|
||||
name: "official endpoint built by the adaptor stays readable",
|
||||
channelType: constant.ChannelTypeVertexAi,
|
||||
origin: "https://us-central1-aiplatform.googleapis.com",
|
||||
message: "request to https://us-central1-aiplatform.googleapis.com/v1/projects failed",
|
||||
want: "request to https://us-central1-aiplatform.googleapis.com/v1/projects failed",
|
||||
},
|
||||
{
|
||||
name: "configured base URL is masked",
|
||||
channelType: constant.ChannelTypeOpenAI,
|
||||
baseURL: "https://relay.private.example",
|
||||
origin: "https://relay.private.example",
|
||||
message: "request to https://relay.private.example/v1/chat/completions failed",
|
||||
want: "request to https://***.example/***/***/*** failed",
|
||||
},
|
||||
{
|
||||
name: "absolute Advanced Custom route is masked",
|
||||
channelType: constant.ChannelTypeAdvancedCustom,
|
||||
origin: "https://route.private.example",
|
||||
message: "dial tcp: lookup route.private.example: no such host",
|
||||
want: "dial tcp: lookup ***.***.example: no such host",
|
||||
},
|
||||
{
|
||||
name: "environment proxy is masked on a preset channel",
|
||||
channelType: constant.ChannelTypeAnthropic,
|
||||
baseURL: "https://api.anthropic.com",
|
||||
proxyEnv: "http://egress.private.invalid:8080",
|
||||
message: "proxyconnect tcp: dial tcp: lookup egress.private.invalid: no such host",
|
||||
want: "proxyconnect tcp: dial tcp: lookup ***.***.invalid: no such host",
|
||||
},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Setenv("HTTPS_PROXY", tc.proxyEnv)
|
||||
t.Setenv("HTTP_PROXY", "")
|
||||
c, _ := gin.CreateTestContext(httptest.NewRecorder())
|
||||
common.SetContextKey(c, constant.ContextKeyChannelType, tc.channelType)
|
||||
common.SetContextKey(c, constant.ContextKeyChannelBaseUrl, tc.baseURL)
|
||||
common.SetContextKey(c, constant.ContextKeyUpstreamRequestOrigin, tc.origin)
|
||||
apiErr := types.NewErrorWithStatusCode(errors.New(tc.message), types.ErrorCodeBadResponseStatusCode, http.StatusBadGateway)
|
||||
|
||||
assert.Equal(t, tc.want, apiErr.ToOpenAIError(common.UpstreamAddresses(c)...).Message)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecideRelayRetryReasons(t *testing.T) {
|
||||
upstream := func(status int) *types.NewAPIError {
|
||||
return types.NewOpenAIError(errors.New("upstream"), types.ErrorCodeBadResponseStatusCode, status)
|
||||
|
||||
@@ -132,7 +132,7 @@ func ChargeViolationFeeIfNeeded(ctx *gin.Context, relayInfo *relaycommon.RelayIn
|
||||
|
||||
useTimeSeconds := time.Now().Unix() - relayInfo.StartTime.Unix()
|
||||
tokenName := ctx.GetString("token_name")
|
||||
oai := apiErr.ToOpenAIError()
|
||||
oai := apiErr.ToOpenAIError(common.UpstreamAddresses(ctx)...)
|
||||
|
||||
other := model.NewLogOther()
|
||||
other.MergePublic(map[string]any{
|
||||
|
||||
Reference in new issue
Block a user