channel_monitor_checker.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649
  1. package service
  2. import (
  3. "bytes"
  4. "context"
  5. "encoding/json"
  6. "errors"
  7. "fmt"
  8. "io"
  9. "net/http"
  10. "net/url"
  11. "regexp"
  12. "strings"
  13. "time"
  14. "github.com/Wei-Shaw/sub2api/internal/pkg/servertiming"
  15. "github.com/tidwall/gjson"
  16. )
  17. // monitorHTTPClient 共享一个 http.Client,避免每次检测重建 transport。
  18. // 自定义 Transport 在 dial 时强制再次校验 IP,防止 DNS rebinding 绕过 validateEndpoint。
  19. var monitorHTTPClient = newSSRFSafeHTTPClient(monitorRequestTimeout)
  20. // monitorPingHTTPClient 用于 endpoint origin 的 HEAD ping,超时更短。
  21. var monitorPingHTTPClient = newSSRFSafeHTTPClient(monitorPingTimeout)
  22. // newSSRFSafeHTTPClient 返回一个使用 safeDialContext 的 http.Client。
  23. // 仅供监控模块对外发起请求使用——所有目标都应是公网 endpoint。
  24. func newSSRFSafeHTTPClient(timeout time.Duration) *http.Client {
  25. tr := &http.Transport{
  26. DialContext: safeDialContext,
  27. ForceAttemptHTTP2: true,
  28. MaxIdleConns: 16,
  29. IdleConnTimeout: monitorIdleConnTimeout,
  30. TLSHandshakeTimeout: monitorTLSHandshakeTimeout,
  31. ResponseHeaderTimeout: monitorResponseHeaderTimeout,
  32. }
  33. return &http.Client{Timeout: timeout, Transport: servertiming.WrapRoundTripper(tr)}
  34. }
  35. // CheckOptions 承载一次检测的自定义入参。
  36. // 所有字段都是可选(零值即等价于"用默认行为")。
  37. type CheckOptions struct {
  38. // APIMode 仅对 OpenAI provider 生效;空串等同 chat_completions。
  39. APIMode string
  40. // ExtraHeaders 用户自定义 HTTP 头(merge 到 adapter 默认 headers,用户优先)。
  41. ExtraHeaders map[string]string
  42. // BodyOverrideMode: off | merge | replace
  43. BodyOverrideMode string
  44. // BodyOverride 在 merge 模式下做浅合并(key 命中黑名单时静默丢弃),
  45. // 在 replace 模式下直接当作完整 body。
  46. BodyOverride map[string]any
  47. }
  48. // runCheckForModel 对单个 (provider, model) 做一次完整检测。
  49. // 不返回 error:所有失败都包装进 CheckResult.Status=error/failed。
  50. //
  51. // opts 承载模板 / 监控快照带来的自定义配置。nil 等同于 "off + 无 extra headers"。
  52. func runCheckForModel(ctx context.Context, provider, endpoint, apiKey, model string, opts *CheckOptions) *CheckResult {
  53. res := &CheckResult{
  54. Model: model,
  55. Status: MonitorStatusError,
  56. CheckedAt: time.Now(),
  57. }
  58. challenge := generateChallenge()
  59. mode := bodyOverrideMode(opts)
  60. start := time.Now()
  61. respText, rawBody, statusCode, err := callProvider(ctx, provider, endpoint, apiKey, model, challenge.Prompt, opts)
  62. latency := time.Since(start)
  63. latencyMs := int(latency / time.Millisecond)
  64. res.LatencyMs = &latencyMs
  65. if err != nil {
  66. res.Status = MonitorStatusError
  67. res.Message = truncateMessage(sanitizeErrorMessage(err.Error()))
  68. return res
  69. }
  70. if statusCode < 200 || statusCode >= 300 {
  71. // 错误路径:用 rawBody 而非 respText(gjson textPath 抽取在错误响应里通常为空,
  72. // 会丢掉真正的上游错误信息,例如 `{"error":{"message":"No available accounts ..."}}`)。
  73. res.Status = MonitorStatusError
  74. bodySnippet := truncateForErrorBody(rawBody)
  75. res.Message = truncateMessage(sanitizeErrorMessage(fmt.Sprintf("upstream HTTP %d: %s", statusCode, bodySnippet)))
  76. return res
  77. }
  78. // Replace 模式:跳过 challenge 校验(用户 body 是静态的,challenge 没法嵌入)。
  79. // 改用「HTTP 2xx + 响应文本(adapter.textPath 抽取)非空」作为 operational 判定。
  80. // 响应文本为空则降级为 failed(视为上游回了 200 但没实际内容)。
  81. if mode == MonitorBodyOverrideModeReplace {
  82. if strings.TrimSpace(respText) == "" {
  83. res.Status = MonitorStatusFailed
  84. res.Message = truncateMessage("replace-mode: upstream returned 2xx with empty text")
  85. return res
  86. }
  87. return finalizeOperationalOrDegraded(res, latency, latencyMs)
  88. }
  89. if !validateChallenge(respText, challenge.Expected) {
  90. res.Status = MonitorStatusFailed
  91. res.Message = truncateMessage(sanitizeErrorMessage(fmt.Sprintf("challenge mismatch (expected %s, got %q)", challenge.Expected, respText)))
  92. return res
  93. }
  94. return finalizeOperationalOrDegraded(res, latency, latencyMs)
  95. }
  96. // finalizeOperationalOrDegraded 负责走到最后一步的 operational/degraded 判定。
  97. // 拆出来是为了让 runCheckForModel 不超过 30 行。
  98. func finalizeOperationalOrDegraded(res *CheckResult, latency time.Duration, latencyMs int) *CheckResult {
  99. if latency >= monitorDegradedThreshold {
  100. res.Status = MonitorStatusDegraded
  101. res.Message = truncateMessage(fmt.Sprintf("slow response: %dms", latencyMs))
  102. return res
  103. }
  104. res.Status = MonitorStatusOperational
  105. return res
  106. }
  107. // bodyOverrideMode 归一取 opts.BodyOverrideMode,nil opts / 空串都视为 off。
  108. func bodyOverrideMode(opts *CheckOptions) string {
  109. if opts == nil || opts.BodyOverrideMode == "" {
  110. return MonitorBodyOverrideModeOff
  111. }
  112. return opts.BodyOverrideMode
  113. }
  114. // pingEndpointOrigin 对 endpoint 的 origin (scheme://host) 发起 HEAD 请求,返回耗时。
  115. // 失败时返回 nil(不影响主状态判定)。
  116. func pingEndpointOrigin(ctx context.Context, endpoint string) *int {
  117. origin, err := extractOrigin(endpoint)
  118. if err != nil || origin == "" {
  119. return nil
  120. }
  121. req, err := http.NewRequestWithContext(ctx, http.MethodHead, origin, nil)
  122. if err != nil {
  123. return nil
  124. }
  125. start := time.Now()
  126. resp, err := monitorPingHTTPClient.Do(req)
  127. if err != nil {
  128. return nil
  129. }
  130. defer func() { _ = resp.Body.Close() }()
  131. _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, monitorPingDiscardMaxBytes))
  132. ms := int(time.Since(start) / time.Millisecond)
  133. return &ms
  134. }
  135. // providerAdapter 描述某个 provider 在 challenge 检测中需要的几件事:
  136. // - 拼出请求路径(含 model 占位)
  137. // - 序列化请求体
  138. // - 构造鉴权头
  139. // - 从响应 JSON 中提取文本(默认按 gjson path;需要时可自定义)
  140. //
  141. // 加新 provider 只需要在 providerAdapters 里增加一个条目,无需触碰 callProvider / validateProvider。
  142. type providerAdapter struct {
  143. buildPath func(model string) string
  144. buildBody func(model, prompt string) ([]byte, error)
  145. buildHeaders func(apiKey string) map[string]string
  146. textPath string // gjson 提取响应文本的 path
  147. extractText func([]byte) string
  148. }
  149. // providerAdapters 全部已支持的 provider。键值即 MonitorProvider* 字符串。
  150. //
  151. //nolint:gochecknoglobals // 适配器表是只读静态数据,初始化后不变更。
  152. var providerAdapters = map[string]providerAdapter{
  153. MonitorProviderOpenAI: providerOpenAIChatAdapter,
  154. MonitorProviderGrok: providerGrokChatAdapter,
  155. // 国产 3 家(配额模式引入):均为 OpenAI 兼容 Chat Completions,
  156. // 仅智谱路径前缀不同(/api/paas/v4/chat/completions)。
  157. MonitorProviderKimi: providerKimiChatAdapter,
  158. MonitorProviderZhipu: providerZhipuChatAdapter,
  159. MonitorProviderDeepseek: providerDeepseekChatAdapter,
  160. MonitorProviderMiniMax: providerMiniMaxChatAdapter,
  161. MonitorProviderAnthropic: {
  162. buildPath: func(string) string { return providerAnthropicPath },
  163. buildBody: func(model, prompt string) ([]byte, error) {
  164. return json.Marshal(map[string]any{
  165. "model": model,
  166. "messages": []map[string]string{{"role": "user", "content": prompt}},
  167. "max_tokens": monitorChallengeMaxTokens,
  168. })
  169. },
  170. buildHeaders: func(apiKey string) map[string]string {
  171. return map[string]string{
  172. "x-api-key": apiKey,
  173. "anthropic-version": monitorAnthropicAPIVersion,
  174. }
  175. },
  176. extractText: extractAnthropicMonitorText,
  177. },
  178. MonitorProviderGemini: {
  179. // Gemini 把 model 名写在 URL path 上:/v1beta/models/{model}:generateContent
  180. buildPath: func(model string) string { return fmt.Sprintf(providerGeminiPathTemplate, model) },
  181. buildBody: func(_, prompt string) ([]byte, error) {
  182. return json.Marshal(map[string]any{
  183. "contents": []map[string]any{
  184. {"role": "user", "parts": []map[string]any{{"text": prompt}}},
  185. },
  186. "generationConfig": map[string]any{"maxOutputTokens": monitorChallengeMaxTokens},
  187. })
  188. },
  189. // 使用 x-goog-api-key header 而不是 ?key= query,避免 *url.Error 把 key 回填到错误日志。
  190. buildHeaders: func(apiKey string) map[string]string {
  191. return map[string]string{"x-goog-api-key": apiKey}
  192. },
  193. textPath: "candidates.0.content.parts.0.text",
  194. },
  195. }
  196. //nolint:gochecknoglobals // 适配器表是只读静态数据,初始化后不变更。
  197. var providerOpenAIChatAdapter = newOpenAICompatibleChatAdapter(providerOpenAIPath)
  198. //nolint:gochecknoglobals // 适配器表是只读静态数据,初始化后不变更。
  199. var providerGrokChatAdapter = newOpenAICompatibleChatAdapter(providerGrokPath)
  200. //nolint:gochecknoglobals // 适配器表是只读静态数据,初始化后不变更。
  201. var providerKimiChatAdapter = newOpenAICompatibleChatAdapter(providerOpenAIPath)
  202. //nolint:gochecknoglobals // 适配器表是只读静态数据,初始化后不变更。
  203. var providerZhipuChatAdapter = newOpenAICompatibleChatAdapter(providerZhipuPath)
  204. //nolint:gochecknoglobals // 适配器表是只读静态数据,初始化后不变更。
  205. var providerDeepseekChatAdapter = newOpenAICompatibleChatAdapter(providerOpenAIPath)
  206. //nolint:gochecknoglobals // 适配器表是只读静态数据,初始化后不变更。
  207. var providerMiniMaxChatAdapter = newOpenAICompatibleChatAdapter(providerOpenAIPath)
  208. func newOpenAICompatibleChatAdapter(path string) providerAdapter {
  209. return providerAdapter{
  210. buildPath: func(string) string { return path },
  211. buildBody: func(model, prompt string) ([]byte, error) {
  212. return json.Marshal(map[string]any{
  213. "model": model,
  214. "messages": []map[string]string{{"role": "user", "content": prompt}},
  215. "max_tokens": monitorChallengeMaxTokens,
  216. "stream": false,
  217. })
  218. },
  219. buildHeaders: func(apiKey string) map[string]string {
  220. return map[string]string{"Authorization": "Bearer " + apiKey}
  221. },
  222. textPath: "choices.0.message.content",
  223. }
  224. }
  225. //nolint:gochecknoglobals // 适配器表是只读静态数据,初始化后不变更。
  226. var providerOpenAIResponsesAdapter = providerAdapter{
  227. buildPath: func(string) string { return providerOpenAIResponsesPath },
  228. buildBody: func(model, prompt string) ([]byte, error) {
  229. return json.Marshal(map[string]any{
  230. "model": model,
  231. "instructions": "You are a channel health-check endpoint. Answer the arithmetic challenge exactly and briefly.",
  232. "input": prompt,
  233. "max_output_tokens": monitorChallengeMaxTokens,
  234. "stream": false,
  235. })
  236. },
  237. buildHeaders: func(apiKey string) map[string]string {
  238. return map[string]string{"Authorization": "Bearer " + apiKey}
  239. },
  240. textPath: "output.0.content.0.text",
  241. }
  242. // providerAdapterFor 按 provider + api_mode 选择具体 adapter。
  243. func providerAdapterFor(provider, apiMode string) (providerAdapter, string, bool) {
  244. if provider == MonitorProviderOpenAI && defaultAPIMode(apiMode) == MonitorAPIModeResponses {
  245. return providerOpenAIResponsesAdapter, MonitorAPIModeResponses, true
  246. }
  247. adapter, ok := providerAdapters[provider]
  248. return adapter, MonitorAPIModeChatCompletions, ok
  249. }
  250. // callProvider 通过 providerAdapters 分发到具体实现。
  251. // opts 承载用户的自定义 headers / body 覆盖(可为 nil)。
  252. //
  253. // 返回值:
  254. // - extractedText: 按 textPath 抽出的成功文本,仅在 status 2xx 时有意义;非 2xx 时通常为空串
  255. // - rawBody: 完整响应体的字符串形式(已被 monitorResponseMaxBytes 截断),用于错误路径保留上游真实回包
  256. // - status: HTTP 状态码
  257. // - err: 网络 / 序列化错误
  258. func callProvider(ctx context.Context, provider, endpoint, apiKey, model, prompt string, opts *CheckOptions) (extractedText, rawBody string, status int, err error) {
  259. requestedAPIMode := checkAPIMode(opts)
  260. if err := validateAPIMode(provider, requestedAPIMode); err != nil {
  261. return "", "", 0, err
  262. }
  263. adapter, apiMode, ok := providerAdapterFor(provider, requestedAPIMode)
  264. if !ok {
  265. return "", "", 0, fmt.Errorf("unsupported provider %q", provider)
  266. }
  267. body, err := buildRequestBody(adapter, provider, apiMode, model, prompt, opts)
  268. if err != nil {
  269. return "", "", 0, err
  270. }
  271. headers := mergeHeaders(adapter.buildHeaders(apiKey), opts)
  272. full := joinURL(endpoint, adapter.buildPath(model))
  273. respBytes, status, err := postRawJSON(ctx, full, body, headers)
  274. if err != nil {
  275. return "", "", status, err
  276. }
  277. if provider == MonitorProviderOpenAI && apiMode == MonitorAPIModeResponses {
  278. return extractOpenAIResponsesText(respBytes), string(respBytes), status, nil
  279. }
  280. return extractMonitorResponseText(adapter, respBytes), string(respBytes), status, nil
  281. }
  282. func extractMonitorResponseText(adapter providerAdapter, respBytes []byte) string {
  283. if adapter.extractText != nil {
  284. return adapter.extractText(respBytes)
  285. }
  286. return gjson.GetBytes(respBytes, adapter.textPath).String()
  287. }
  288. func extractAnthropicMonitorText(respBytes []byte) string {
  289. content := gjson.GetBytes(respBytes, "content")
  290. if !content.IsArray() {
  291. return ""
  292. }
  293. parts := make([]string, 0, 1)
  294. content.ForEach(func(_, item gjson.Result) bool {
  295. if item.Get("type").String() != "text" {
  296. return true
  297. }
  298. text := strings.TrimSpace(item.Get("text").String())
  299. if text != "" {
  300. parts = append(parts, text)
  301. }
  302. return true
  303. })
  304. return strings.Join(parts, "\n")
  305. }
  306. // extractOpenAIResponsesText 聚合 Responses API 的最终 assistant 文本。
  307. // Responses 的 output 数组顺序由模型决定:reasoning / tool-call item 可能排在 message 前面,
  308. // 因此不能假设文本永远在 output.0.content.0.text。
  309. func extractOpenAIResponsesText(respBytes []byte) string {
  310. if text := gjson.GetBytes(respBytes, "output_text").String(); strings.TrimSpace(text) != "" {
  311. return text
  312. }
  313. var texts []string
  314. outputs := gjson.GetBytes(respBytes, "output")
  315. if outputs.IsArray() {
  316. outputs.ForEach(func(_, output gjson.Result) bool {
  317. outputType := output.Get("type").String()
  318. if outputType != "" && outputType != "message" {
  319. return true
  320. }
  321. content := output.Get("content")
  322. if !content.IsArray() {
  323. return true
  324. }
  325. content.ForEach(func(_, block gjson.Result) bool {
  326. blockType := block.Get("type").String()
  327. if blockType != "" && blockType != "output_text" {
  328. return true
  329. }
  330. if text := block.Get("text").String(); strings.TrimSpace(text) != "" {
  331. texts = append(texts, text)
  332. }
  333. return true
  334. })
  335. return true
  336. })
  337. }
  338. if len(texts) > 0 {
  339. return strings.Join(texts, "")
  340. }
  341. return gjson.GetBytes(respBytes, providerOpenAIResponsesAdapter.textPath).String()
  342. }
  343. // mergeHeaders 把用户自定义 headers 合并到 adapter 默认 headers 上。
  344. // 用户值覆盖默认;命中黑名单(hop-by-hop / 由 http.Client 自管的)的 key 静默丢弃。
  345. func mergeHeaders(base map[string]string, opts *CheckOptions) map[string]string {
  346. if opts == nil || len(opts.ExtraHeaders) == 0 {
  347. return base
  348. }
  349. out := make(map[string]string, len(base)+len(opts.ExtraHeaders))
  350. for k, v := range base {
  351. out[k] = v
  352. }
  353. for k, v := range opts.ExtraHeaders {
  354. if IsForbiddenHeaderName(k) {
  355. continue
  356. }
  357. out[k] = v
  358. }
  359. return out
  360. }
  361. // buildRequestBody 根据 body_override_mode 构造请求 body。
  362. //
  363. // - off: adapter 默认 body
  364. // - merge: adapter 默认 body 与 BodyOverride 浅合并;BodyOverride 中命中
  365. // bodyMergeKeyDenyList[provider] 的 key 会被静默丢弃,避免破坏 challenge / model 路由
  366. // - replace: 直接 marshal BodyOverride 作为完整 body
  367. //
  368. // 任何 mode 返回的 []byte 都已经是合法 JSON,可直接送入 postRawJSON。
  369. func buildRequestBody(adapter providerAdapter, provider, apiMode, model, prompt string, opts *CheckOptions) ([]byte, error) {
  370. mode := bodyOverrideMode(opts)
  371. if mode == MonitorBodyOverrideModeReplace {
  372. if opts == nil || len(opts.BodyOverride) == 0 {
  373. return nil, fmt.Errorf("replace mode: body_override is empty")
  374. }
  375. if err := validateReplaceRequestBody(provider, apiMode, opts.BodyOverride); err != nil {
  376. return nil, err
  377. }
  378. body, err := json.Marshal(opts.BodyOverride)
  379. if err != nil {
  380. return nil, fmt.Errorf("marshal body_override (replace): %w", err)
  381. }
  382. return body, nil
  383. }
  384. defaultBody, err := adapter.buildBody(model, prompt)
  385. if err != nil {
  386. return nil, fmt.Errorf("marshal default body: %w", err)
  387. }
  388. if mode != MonitorBodyOverrideModeMerge || opts == nil || len(opts.BodyOverride) == 0 {
  389. return defaultBody, nil
  390. }
  391. var defaultMap map[string]any
  392. if err := json.Unmarshal(defaultBody, &defaultMap); err != nil {
  393. return nil, fmt.Errorf("unmarshal default body for merge: %w", err)
  394. }
  395. deny := bodyMergeKeyDenyList[bodyMergeDenyKey(provider, apiMode)]
  396. for k, v := range opts.BodyOverride {
  397. if deny[k] {
  398. continue
  399. }
  400. defaultMap[k] = v
  401. }
  402. merged, err := json.Marshal(defaultMap)
  403. if err != nil {
  404. return nil, fmt.Errorf("marshal merged body: %w", err)
  405. }
  406. return merged, nil
  407. }
  408. // bodyMergeKeyDenyList 在 merge 模式下,禁止用户覆盖这些 provider-specific 的关键字段。
  409. // 思路抄 check-cx 的 EXCLUDED_METADATA_KEYS:保护 challenge / model 路由不被用户误伤。
  410. // 用户想动这些字段就用 replace 模式(已知会跳 challenge 校验)。
  411. //
  412. //nolint:gochecknoglobals // 静态查表,初始化后不变。
  413. var bodyMergeKeyDenyList = map[string]map[string]bool{
  414. MonitorProviderOpenAI + ":" + MonitorAPIModeChatCompletions: {"model": true, "messages": true, "stream": true},
  415. MonitorProviderOpenAI + ":" + MonitorAPIModeResponses: {"model": true, "instructions": true, "input": true, "stream": true},
  416. MonitorProviderGrok: {"model": true, "messages": true, "stream": true},
  417. MonitorProviderAnthropic: {"model": true, "messages": true},
  418. MonitorProviderGemini: {"contents": true},
  419. // 国产 3 家与 OpenAI Chat Completions 同构。
  420. MonitorProviderKimi: {"model": true, "messages": true, "stream": true},
  421. MonitorProviderZhipu: {"model": true, "messages": true, "stream": true},
  422. MonitorProviderDeepseek: {"model": true, "messages": true, "stream": true},
  423. MonitorProviderMiniMax: {"model": true, "messages": true, "stream": true},
  424. }
  425. func checkAPIMode(opts *CheckOptions) string {
  426. if opts == nil {
  427. return MonitorAPIModeChatCompletions
  428. }
  429. return defaultAPIMode(opts.APIMode)
  430. }
  431. func bodyMergeDenyKey(provider, apiMode string) string {
  432. if provider == MonitorProviderOpenAI {
  433. return provider + ":" + defaultAPIMode(apiMode)
  434. }
  435. return provider
  436. }
  437. // isOpenAICompatibleChatProvider 该 provider 的探活请求是否为 OpenAI Chat
  438. // Completions 同构(replace 模式的 body 校验按 messages 必填处理)。
  439. func isOpenAICompatibleChatProvider(provider string) bool {
  440. switch provider {
  441. case MonitorProviderOpenAI, MonitorProviderGrok,
  442. MonitorProviderKimi, MonitorProviderZhipu, MonitorProviderDeepseek, MonitorProviderMiniMax:
  443. return true
  444. default:
  445. return false
  446. }
  447. }
  448. func validateReplaceRequestBody(provider, apiMode string, body map[string]any) error {
  449. if !isOpenAICompatibleChatProvider(provider) {
  450. return nil
  451. }
  452. switch defaultAPIMode(apiMode) {
  453. case MonitorAPIModeResponses:
  454. if strings.TrimSpace(stringFromAny(body["instructions"])) == "" || !hasNonEmptyBodyValue(body["input"]) {
  455. return fmt.Errorf("replace mode responses body: instructions and input are required")
  456. }
  457. case MonitorAPIModeChatCompletions:
  458. if !hasNonEmptyBodyValue(body["messages"]) {
  459. return fmt.Errorf("replace mode chat_completions body: messages are required")
  460. }
  461. }
  462. return nil
  463. }
  464. func stringFromAny(v any) string {
  465. s, _ := v.(string)
  466. return s
  467. }
  468. func hasNonEmptyBodyValue(v any) bool {
  469. switch val := v.(type) {
  470. case nil:
  471. return false
  472. case string:
  473. return strings.TrimSpace(val) != ""
  474. case []any:
  475. return len(val) > 0
  476. case []map[string]any:
  477. return len(val) > 0
  478. case []map[string]string:
  479. return len(val) > 0
  480. default:
  481. return true
  482. }
  483. }
  484. // postRawJSON 发送 POST + 已序列化好的 JSON 字节,限制响应体大小,返回响应字节、HTTP status、错误。
  485. // adapter 自行 marshal 是为了精确控制字段顺序与类型,所以这里直接收 []byte 而不是 any。
  486. func postRawJSON(ctx context.Context, fullURL string, payload []byte, headers map[string]string) ([]byte, int, error) {
  487. req, err := http.NewRequestWithContext(ctx, http.MethodPost, fullURL, bytes.NewReader(payload))
  488. if err != nil {
  489. return nil, 0, fmt.Errorf("build request: %w", err)
  490. }
  491. req.Header.Set("Content-Type", "application/json")
  492. req.Header.Set("Accept", "application/json")
  493. for k, v := range headers {
  494. req.Header.Set(k, v)
  495. }
  496. resp, err := monitorHTTPClient.Do(req)
  497. if err != nil {
  498. return nil, 0, fmt.Errorf("do request: %w", err)
  499. }
  500. defer func() { _ = resp.Body.Close() }()
  501. respBody, err := io.ReadAll(io.LimitReader(resp.Body, monitorResponseMaxBytes))
  502. if err != nil {
  503. return nil, resp.StatusCode, fmt.Errorf("read body: %w", err)
  504. }
  505. return respBody, resp.StatusCode, nil
  506. }
  507. // joinURL 把 base origin 与 path 拼成完整 URL。
  508. // 容忍 base 末尾有/无斜杠,path 必带前导斜杠。
  509. func joinURL(base, path string) string {
  510. base = strings.TrimRight(base, "/")
  511. if !strings.HasPrefix(path, "/") {
  512. path = "/" + path
  513. }
  514. return base + path
  515. }
  516. // extractOrigin 从一个 endpoint URL 中提取 scheme://host[:port] 部分。
  517. func extractOrigin(endpoint string) (string, error) {
  518. u, err := url.Parse(endpoint)
  519. if err != nil {
  520. return "", err
  521. }
  522. if u.Scheme == "" || u.Host == "" {
  523. return "", errors.New("endpoint missing scheme or host")
  524. }
  525. return u.Scheme + "://" + u.Host, nil
  526. }
  527. // monitorSensitiveQueryParamRegex 匹配 URL query 中可能泄露凭证的参数:
  528. // key / api_key / api-key / access_token / token / authorization / x-api-key。
  529. // 大小写不敏感,匹配 `?name=value` 或 `&name=value` 形式(value 截到 & 或字符串末尾)。
  530. var monitorSensitiveQueryParamRegex = regexp.MustCompile(`(?i)([?&](?:key|api[_-]?key|access[_-]?token|token|authorization|x-api-key)=)[^&\s"']+`)
  531. // monitorAPIKeyPatterns 匹配常见 provider 的 API key 字面量。
  532. // 顺序敏感:sk-ant- 必须放在 sk- 之前,否则会被通用 sk- 模式先消费。
  533. var monitorAPIKeyPatterns = []struct {
  534. pattern *regexp.Regexp
  535. replace string
  536. }{
  537. // Anthropic(带前缀,必须先匹配):sk-ant-xxxxxxx
  538. {regexp.MustCompile(`sk-ant-[A-Za-z0-9_-]{20,}`), "sk-ant-***REDACTED***"},
  539. // OpenAI / Anthropic 通用 sk-: sk-xxxxxxx
  540. {regexp.MustCompile(`sk-[A-Za-z0-9-]{20,}`), "sk-***REDACTED***"},
  541. // xAI API Key:xai-xxxxxxx
  542. {regexp.MustCompile(`xai-[A-Za-z0-9_-]{6,}`), "xai-***REDACTED***"},
  543. // Gemini / Google API Key:固定前缀 + 35 位
  544. {regexp.MustCompile(`AIza[A-Za-z0-9_-]{35}`), "AIza***REDACTED***"},
  545. // JWT 三段式(Bearer 后常出现):eyJxxx.eyJxxx.signature
  546. {regexp.MustCompile(`eyJ[A-Za-z0-9_-]{8,}\.eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}`), "eyJ***REDACTED.JWT***"},
  547. }
  548. // sanitizeErrorMessage 擦除错误/响应文本中可能泄露的 API key。
  549. // 处理两类来源:
  550. // 1. URL query 中的 ?key= / ?api_key= 等(Go *url.Error 会回填完整 URL)
  551. // 2. 上游 HTTP body 文本里直接出现的 sk-* / xai-* / AIza* / JWT 等密钥碎片
  552. //
  553. // 注意:与 gemini_messages_compat_service.go 的 sanitizeUpstreamErrorMessage 关注点类似但参数集更广,
  554. // 监控模块独立维护,避免互相耦合。
  555. func sanitizeErrorMessage(msg string) string {
  556. if msg == "" {
  557. return msg
  558. }
  559. msg = monitorSensitiveQueryParamRegex.ReplaceAllString(msg, `${1}REDACTED`)
  560. for _, p := range monitorAPIKeyPatterns {
  561. msg = p.pattern.ReplaceAllString(msg, p.replace)
  562. }
  563. return msg
  564. }
  565. // truncateMessage 把消息按 monitorMessageMaxBytes 截断,避免 DB 列溢出与日志过长。
  566. func truncateMessage(msg string) string {
  567. if len(msg) <= monitorMessageMaxBytes {
  568. return msg
  569. }
  570. const ellipsis = "...(truncated)"
  571. cutoff := monitorMessageMaxBytes - len(ellipsis)
  572. if cutoff < 0 {
  573. cutoff = 0
  574. }
  575. return msg[:cutoff] + ellipsis
  576. }
  577. // truncateForErrorBody 把上游错误响应 body 压到 monitorErrorBodySnippetMaxBytes 以内,
  578. // 并顺手把连续空白折成一个空格:上游 HTML 错误页常含大量缩进/换行,保留会浪费预算。
  579. // 被 truncateMessage 做最终总截断兜底,所以这里只负责 body 自身的精简。
  580. func truncateForErrorBody(body string) string {
  581. body = strings.Join(strings.Fields(body), " ")
  582. if len(body) <= monitorErrorBodySnippetMaxBytes {
  583. return body
  584. }
  585. const ellipsis = "...(body truncated)"
  586. cutoff := monitorErrorBodySnippetMaxBytes - len(ellipsis)
  587. if cutoff < 0 {
  588. cutoff = 0
  589. }
  590. return body[:cutoff] + ellipsis
  591. }