http_upstream_test.go 40 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035
  1. package repository
  2. import (
  3. "bytes"
  4. "encoding/binary"
  5. "errors"
  6. "fmt"
  7. "io"
  8. "net"
  9. "net/http"
  10. "net/http/httptest"
  11. "net/url"
  12. "strings"
  13. "sync/atomic"
  14. "testing"
  15. "time"
  16. "github.com/Wei-Shaw/sub2api/internal/config"
  17. "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
  18. "github.com/Wei-Shaw/sub2api/internal/pkg/xai"
  19. "github.com/Wei-Shaw/sub2api/internal/service"
  20. "github.com/stretchr/testify/require"
  21. "github.com/stretchr/testify/suite"
  22. )
  23. func TestHTTPUpstreamDoCanDisableRedirectsPerRequest(t *testing.T) {
  24. var redirectedCalls atomic.Int64
  25. target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
  26. redirectedCalls.Add(1)
  27. w.WriteHeader(http.StatusOK)
  28. }))
  29. t.Cleanup(target.Close)
  30. redirector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  31. http.Redirect(w, r, target.URL, http.StatusFound)
  32. }))
  33. t.Cleanup(redirector.Close)
  34. upstream := NewHTTPUpstream(nil)
  35. req, err := http.NewRequestWithContext(
  36. service.WithHTTPUpstreamRedirectsDisabled(t.Context()),
  37. http.MethodGet,
  38. redirector.URL,
  39. nil,
  40. )
  41. require.NoError(t, err)
  42. resp, err := upstream.Do(req, "", 1, 1)
  43. require.NoError(t, err)
  44. require.Equal(t, http.StatusFound, resp.StatusCode)
  45. require.NoError(t, resp.Body.Close())
  46. require.Zero(t, redirectedCalls.Load())
  47. }
  48. func TestHTTPUpstreamDoWithTLSRejectsPlainHTTPBeforeHTTPProxy(t *testing.T) {
  49. var upstreamCalls atomic.Int64
  50. upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
  51. upstreamCalls.Add(1)
  52. w.WriteHeader(http.StatusTeapot)
  53. }))
  54. t.Cleanup(upstream.Close)
  55. var proxyCalls atomic.Int64
  56. proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
  57. proxyCalls.Add(1)
  58. w.WriteHeader(http.StatusNoContent)
  59. }))
  60. t.Cleanup(proxy.Close)
  61. req, err := http.NewRequest(http.MethodGet, upstream.URL, nil)
  62. require.NoError(t, err)
  63. client := NewHTTPUpstream(nil)
  64. resp, err := client.DoWithTLS(req, proxy.URL, 41, 1, &tlsfingerprint.Profile{Name: "unused-for-http"})
  65. require.ErrorContains(t, err, "requires HTTPS")
  66. require.Nil(t, resp)
  67. require.Zero(t, proxyCalls.Load())
  68. require.Zero(t, upstreamCalls.Load())
  69. }
  70. func TestHTTPUpstreamDoWithTLSRejectsPlainHTTPBeforeSOCKSProxy(t *testing.T) {
  71. var upstreamCalls atomic.Int64
  72. upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
  73. upstreamCalls.Add(1)
  74. w.WriteHeader(http.StatusNoContent)
  75. }))
  76. t.Cleanup(upstream.Close)
  77. proxyURL, proxyCalls := startTestSOCKS5Proxy(t)
  78. req, err := http.NewRequest(http.MethodGet, upstream.URL, nil)
  79. require.NoError(t, err)
  80. client := NewHTTPUpstream(nil)
  81. resp, err := client.DoWithTLS(req, proxyURL, 42, 1, &tlsfingerprint.Profile{Name: "unused-for-http"})
  82. require.ErrorContains(t, err, "requires HTTPS")
  83. require.Nil(t, resp)
  84. require.Zero(t, proxyCalls.Load())
  85. require.Zero(t, upstreamCalls.Load())
  86. }
  87. func TestTLSFingerprintHTTPSProxyUsesFingerprintDialer(t *testing.T) {
  88. proxyURL, err := url.Parse("https://user:pass@proxy.example:8443")
  89. require.NoError(t, err)
  90. transport, err := buildUpstreamTransportWithTLSFingerprint(poolSettings{}, proxyURL, &tlsfingerprint.Profile{Name: "test"})
  91. require.NoError(t, err)
  92. require.Nil(t, transport.Proxy, "CONNECT must be handled by the verified fingerprint dialer")
  93. require.NotNil(t, transport.DialTLSContext)
  94. }
  95. func startTestSOCKS5Proxy(t *testing.T) (string, *atomic.Int64) {
  96. t.Helper()
  97. listener, err := net.Listen("tcp", "127.0.0.1:0")
  98. require.NoError(t, err)
  99. t.Cleanup(func() { _ = listener.Close() })
  100. calls := &atomic.Int64{}
  101. go func() {
  102. for {
  103. conn, err := listener.Accept()
  104. if err != nil {
  105. return
  106. }
  107. calls.Add(1)
  108. go serveTestSOCKS5Conn(conn)
  109. }
  110. }()
  111. return "socks5h://" + listener.Addr().String(), calls
  112. }
  113. func serveTestSOCKS5Conn(client net.Conn) {
  114. defer func() { _ = client.Close() }()
  115. header := make([]byte, 2)
  116. if _, err := io.ReadFull(client, header); err != nil || header[0] != 5 {
  117. return
  118. }
  119. methods := make([]byte, int(header[1]))
  120. if _, err := io.ReadFull(client, methods); err != nil {
  121. return
  122. }
  123. if _, err := client.Write([]byte{5, 0}); err != nil {
  124. return
  125. }
  126. request := make([]byte, 4)
  127. if _, err := io.ReadFull(client, request); err != nil || request[0] != 5 || request[1] != 1 {
  128. return
  129. }
  130. var host string
  131. switch request[3] {
  132. case 1:
  133. address := make([]byte, net.IPv4len)
  134. if _, err := io.ReadFull(client, address); err != nil {
  135. return
  136. }
  137. host = net.IP(address).String()
  138. case 3:
  139. length := make([]byte, 1)
  140. if _, err := io.ReadFull(client, length); err != nil {
  141. return
  142. }
  143. address := make([]byte, int(length[0]))
  144. if _, err := io.ReadFull(client, address); err != nil {
  145. return
  146. }
  147. host = string(address)
  148. case 4:
  149. address := make([]byte, net.IPv6len)
  150. if _, err := io.ReadFull(client, address); err != nil {
  151. return
  152. }
  153. host = net.IP(address).String()
  154. default:
  155. return
  156. }
  157. portBytes := make([]byte, 2)
  158. if _, err := io.ReadFull(client, portBytes); err != nil {
  159. return
  160. }
  161. target, err := net.Dial("tcp", net.JoinHostPort(host, fmt.Sprintf("%d", binary.BigEndian.Uint16(portBytes))))
  162. if err != nil {
  163. _, _ = client.Write([]byte{5, 1, 0, 1, 0, 0, 0, 0, 0, 0})
  164. return
  165. }
  166. defer func() { _ = target.Close() }()
  167. if _, err := client.Write([]byte{5, 0, 0, 1, 0, 0, 0, 0, 0, 0}); err != nil {
  168. return
  169. }
  170. go func() { _, _ = io.Copy(target, client); _ = target.Close() }()
  171. _, _ = io.Copy(client, target)
  172. }
  173. func TestHTTPUpstreamDoAppliesGrokCLIIdentityBeforeOAuthRoundTrip(t *testing.T) {
  174. t.Setenv("XAI_GROK_CLI_VERSION", "")
  175. for _, endpoint := range []string{"responses", "chat/completions"} {
  176. t.Run(endpoint, func(t *testing.T) {
  177. upstream := NewHTTPUpstream(nil)
  178. svc, ok := upstream.(*httpUpstreamService)
  179. require.True(t, ok)
  180. const accountID int64 = 4084
  181. isolation := svc.getIsolationMode()
  182. profile := service.HTTPUpstreamProfileDefault
  183. proxyKey := directProxyKey
  184. protocolMode := svc.resolveProtocolMode(profile, proxyKey, nil)
  185. settings := svc.resolvePoolSettings(isolation, 1)
  186. settings = svc.applyProfilePoolSettings(settings, profile)
  187. cacheKey := buildCacheKey(isolation, proxyKey, accountID, protocolMode)
  188. var capturedHeaders http.Header
  189. svc.clients[cacheKey] = &upstreamClientEntry{
  190. client: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
  191. capturedHeaders = req.Header.Clone()
  192. statusCode := http.StatusOK
  193. if req.Header.Get("X-XAI-Token-Auth") != "xai-grok-cli" {
  194. statusCode = http.StatusForbidden
  195. }
  196. return &http.Response{
  197. StatusCode: statusCode,
  198. Header: make(http.Header),
  199. Body: http.NoBody,
  200. Request: req,
  201. }, nil
  202. })},
  203. proxyKey: proxyKey,
  204. poolKey: buildPoolKey(settings, protocolMode),
  205. protocolMode: protocolMode,
  206. }
  207. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/"+endpoint, nil)
  208. require.NoError(t, err)
  209. req.Header.Set("User-Agent", "legacy-client/1.0")
  210. resp, err := svc.Do(req, "", accountID, 1)
  211. require.NoError(t, err)
  212. require.Equal(t, http.StatusOK, resp.StatusCode)
  213. require.NoError(t, resp.Body.Close())
  214. require.Equal(t, xai.CLIClientVersion, capturedHeaders.Get("x-grok-client-version"))
  215. require.Equal(t, "xai-grok-cli", capturedHeaders.Get("X-XAI-Token-Auth"))
  216. require.Equal(t, xai.CLIUserAgent(xai.CLIClientVersion), capturedHeaders.Get("User-Agent"))
  217. })
  218. }
  219. }
  220. func TestHTTPUpstreamDoFallsBackToOfficialGrokAPIOnCLIAccessDenied(t *testing.T) {
  221. upstream := NewHTTPUpstream(nil)
  222. svc, ok := upstream.(*httpUpstreamService)
  223. require.True(t, ok)
  224. const accountID int64 = 4421
  225. isolation := svc.getIsolationMode()
  226. profile := service.HTTPUpstreamProfileDefault
  227. proxyKey := directProxyKey
  228. protocolMode := svc.resolveProtocolMode(profile, proxyKey, nil)
  229. settings := svc.resolvePoolSettings(isolation, 1)
  230. settings = svc.applyProfilePoolSettings(settings, profile)
  231. cacheKey := buildCacheKey(isolation, proxyKey, accountID, protocolMode)
  232. payload := []byte(`{"model":"grok-4.5","input":"hello"}`)
  233. var calls int
  234. var fallbackBody []byte
  235. var fallbackHeaders http.Header
  236. svc.clients[cacheKey] = &upstreamClientEntry{
  237. client: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
  238. calls++
  239. body, err := io.ReadAll(req.Body)
  240. require.NoError(t, err)
  241. if calls == 1 {
  242. require.Equal(t, grokCLIProxyHost, req.URL.Hostname())
  243. require.Equal(t, "xai-grok-cli", req.Header.Get("X-XAI-Token-Auth"))
  244. return &http.Response{
  245. StatusCode: http.StatusForbidden,
  246. Header: make(http.Header),
  247. Body: io.NopCloser(strings.NewReader(`{"error":"Access denied"}`)),
  248. Request: req,
  249. }, nil
  250. }
  251. fallbackBody = body
  252. fallbackHeaders = req.Header.Clone()
  253. require.Equal(t, grokOfficialAPIHost, req.URL.Hostname())
  254. require.Equal(t, "/v1/responses", req.URL.Path)
  255. return &http.Response{
  256. StatusCode: http.StatusOK,
  257. Header: make(http.Header),
  258. Body: io.NopCloser(strings.NewReader(`{"id":"response-ok"}`)),
  259. Request: req,
  260. }, nil
  261. })},
  262. proxyKey: proxyKey,
  263. poolKey: buildPoolKey(settings, protocolMode),
  264. protocolMode: protocolMode,
  265. }
  266. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/responses", bytes.NewReader(payload))
  267. require.NoError(t, err)
  268. req.Header.Set("Authorization", "Bearer oauth-token")
  269. resp, err := svc.Do(req, "", accountID, 1)
  270. require.NoError(t, err)
  271. require.Equal(t, http.StatusOK, resp.StatusCode)
  272. responseBody, err := io.ReadAll(resp.Body)
  273. require.NoError(t, err)
  274. require.NoError(t, resp.Body.Close())
  275. require.JSONEq(t, `{"id":"response-ok"}`, string(responseBody))
  276. require.Equal(t, 2, calls)
  277. require.Equal(t, payload, fallbackBody)
  278. require.Equal(t, "Bearer oauth-token", fallbackHeaders.Get("Authorization"))
  279. require.Empty(t, fallbackHeaders.Get("X-XAI-Token-Auth"))
  280. require.Empty(t, fallbackHeaders.Get("x-grok-client-version"))
  281. require.Empty(t, fallbackHeaders.Get("User-Agent"))
  282. }
  283. func TestGrokAccessDeniedFallbackRecognizesChatEndpointPermissionDenied(t *testing.T) {
  284. var hosts []string
  285. transport := &grokAccessDeniedFallbackTransport{
  286. base: roundTripFunc(func(req *http.Request) (*http.Response, error) {
  287. hosts = append(hosts, req.URL.Hostname())
  288. if req.URL.Hostname() == grokCLIProxyHost {
  289. return &http.Response{
  290. StatusCode: http.StatusForbidden,
  291. Header: make(http.Header),
  292. Body: io.NopCloser(strings.NewReader(
  293. `{"code":"permission_denied","error":"Access to the chat endpoint is denied. Please ensure you're using the correct credentials. If you believe this is a mistake, please contact support."}`,
  294. )),
  295. Request: req,
  296. }, nil
  297. }
  298. return &http.Response{
  299. StatusCode: http.StatusOK,
  300. Header: make(http.Header),
  301. Body: io.NopCloser(strings.NewReader(`{"id":"response-ok"}`)),
  302. Request: req,
  303. }, nil
  304. }),
  305. }
  306. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/responses", strings.NewReader(`{"model":"grok-4.5"}`))
  307. require.NoError(t, err)
  308. req.Header.Set("Authorization", "Bearer oauth-token")
  309. req.Header.Set("X-XAI-Token-Auth", "xai-grok-cli")
  310. resp, err := transport.RoundTrip(req)
  311. require.NoError(t, err)
  312. require.Equal(t, http.StatusOK, resp.StatusCode)
  313. require.NoError(t, resp.Body.Close())
  314. require.Equal(t, []string{grokCLIProxyHost, grokOfficialAPIHost}, hosts)
  315. }
  316. func TestIsGrokCLICompatibilityAccessDenied(t *testing.T) {
  317. tests := []struct {
  318. name string
  319. body string
  320. want bool
  321. }{
  322. {name: "legacy compatibility wording", body: `{"error":"Access denied"}`, want: true},
  323. {
  324. name: "observed chat endpoint permission denial",
  325. body: `{"code":"permission_denied","error":"Access to the chat endpoint is denied. Please ensure you're using the correct credentials. If you believe this is a mistake, please contact support."}`,
  326. want: true,
  327. },
  328. {
  329. name: "entitlement denial using the same broad terms",
  330. body: `{"code":"permission_denied","error":"Access to the chat endpoint is denied because a subscription is required"}`,
  331. want: false,
  332. },
  333. {
  334. name: "different permission denied endpoint",
  335. body: `{"code":"permission_denied","error":"Access to the billing endpoint is denied."}`,
  336. want: false,
  337. },
  338. {
  339. name: "wrong structured error code",
  340. body: `{"code":"subscription_required","error":"Access to the chat endpoint is denied. Please ensure you're using the correct credentials. If you believe this is a mistake, please contact support."}`,
  341. want: false,
  342. },
  343. {name: "malformed response", body: `permission_denied: chat endpoint denied`, want: false},
  344. }
  345. for _, tt := range tests {
  346. t.Run(tt.name, func(t *testing.T) {
  347. require.Equal(t, tt.want, isGrokCLICompatibilityAccessDenied([]byte(tt.body)))
  348. })
  349. }
  350. }
  351. func TestIsGrokCLIAccessDeniedFallbackCandidateRequiresAuthenticatedReplayableCLI403(t *testing.T) {
  352. newRequest := func() *http.Request {
  353. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/responses", strings.NewReader(`{"model":"grok-4.5"}`))
  354. require.NoError(t, err)
  355. req.Header.Set("Authorization", "Bearer oauth-token")
  356. req.Header.Set("X-XAI-Token-Auth", "xai-grok-cli")
  357. return req
  358. }
  359. newResponse := func() *http.Response { return &http.Response{StatusCode: http.StatusForbidden} }
  360. t.Run("valid candidate", func(t *testing.T) {
  361. require.True(t, isGrokCLIAccessDeniedFallbackCandidate(newRequest(), newResponse()))
  362. })
  363. t.Run("non CLI host", func(t *testing.T) {
  364. req := newRequest()
  365. req.URL.Host = "api.x.ai"
  366. require.False(t, isGrokCLIAccessDeniedFallbackCandidate(req, newResponse()))
  367. })
  368. t.Run("missing CLI identity", func(t *testing.T) {
  369. req := newRequest()
  370. req.Header.Del("X-XAI-Token-Auth")
  371. require.False(t, isGrokCLIAccessDeniedFallbackCandidate(req, newResponse()))
  372. })
  373. t.Run("missing bearer authentication", func(t *testing.T) {
  374. req := newRequest()
  375. req.Header.Del("Authorization")
  376. require.False(t, isGrokCLIAccessDeniedFallbackCandidate(req, newResponse()))
  377. })
  378. t.Run("non forbidden response", func(t *testing.T) {
  379. resp := newResponse()
  380. resp.StatusCode = http.StatusUnauthorized
  381. require.False(t, isGrokCLIAccessDeniedFallbackCandidate(newRequest(), resp))
  382. })
  383. t.Run("non replayable request", func(t *testing.T) {
  384. req := newRequest()
  385. req.GetBody = nil
  386. require.False(t, isGrokCLIAccessDeniedFallbackCandidate(req, newResponse()))
  387. })
  388. }
  389. func TestHTTPUpstreamDoDoesNotFallbackForGrokEntitlementDenial(t *testing.T) {
  390. transport := &grokAccessDeniedFallbackTransport{
  391. base: roundTripFunc(func(req *http.Request) (*http.Response, error) {
  392. return &http.Response{
  393. StatusCode: http.StatusForbidden,
  394. Header: make(http.Header),
  395. Body: io.NopCloser(strings.NewReader(`{"error":"subscription required"}`)),
  396. Request: req,
  397. }, nil
  398. }),
  399. }
  400. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/responses", strings.NewReader(`{"model":"grok-4.5"}`))
  401. require.NoError(t, err)
  402. req.Header.Set("Authorization", "Bearer oauth-token")
  403. req.Header.Set("X-XAI-Token-Auth", "xai-grok-cli")
  404. resp, err := transport.RoundTrip(req)
  405. require.NoError(t, err)
  406. require.Equal(t, http.StatusForbidden, resp.StatusCode)
  407. body, err := io.ReadAll(resp.Body)
  408. require.NoError(t, err)
  409. require.NoError(t, resp.Body.Close())
  410. require.JSONEq(t, `{"error":"subscription required"}`, string(body))
  411. }
  412. func TestApplyGrokCLIProxyHeaders(t *testing.T) {
  413. t.Run("uses pinned stable version for the CLI proxy", func(t *testing.T) {
  414. t.Setenv("XAI_GROK_CLI_VERSION", "")
  415. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/responses", nil)
  416. require.NoError(t, err)
  417. req.Header.Set("User-Agent", "legacy-client/1.0")
  418. applyGrokCLIProxyHeaders(req)
  419. require.Equal(t, xai.CLIClientVersion, req.Header.Get("x-grok-client-version"))
  420. require.Equal(t, "xai-grok-cli", req.Header.Get("X-XAI-Token-Auth"))
  421. require.Equal(t, xai.CLIUserAgent(xai.CLIClientVersion), req.Header.Get("User-Agent"))
  422. })
  423. t.Run("accepts a valid operator override", func(t *testing.T) {
  424. t.Setenv("XAI_GROK_CLI_VERSION", "0.2.121-alpha.1")
  425. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/chat/completions", nil)
  426. require.NoError(t, err)
  427. applyGrokCLIProxyHeaders(req)
  428. require.Equal(t, "0.2.121-alpha.1", req.Header.Get("x-grok-client-version"))
  429. require.Equal(t, xai.CLIUserAgent("0.2.121-alpha.1"), req.Header.Get("User-Agent"))
  430. })
  431. t.Run("rejects an unsafe override", func(t *testing.T) {
  432. t.Setenv("XAI_GROK_CLI_VERSION", "0.2.121\r\nX-Injected: true")
  433. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/responses", nil)
  434. require.NoError(t, err)
  435. applyGrokCLIProxyHeaders(req)
  436. require.Equal(t, xai.CLIClientVersion, req.Header.Get("x-grok-client-version"))
  437. require.Empty(t, req.Header.Get("X-Injected"))
  438. })
  439. t.Run("rejects an override below the supported minimum", func(t *testing.T) {
  440. t.Setenv("XAI_GROK_CLI_VERSION", "0.2.119")
  441. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/responses", nil)
  442. require.NoError(t, err)
  443. applyGrokCLIProxyHeaders(req)
  444. require.Equal(t, xai.CLIClientVersion, req.Header.Get("x-grok-client-version"))
  445. require.Equal(t, xai.CLIUserAgent(xai.CLIClientVersion), req.Header.Get("User-Agent"))
  446. })
  447. t.Run("rejects a prerelease override at the minimum version", func(t *testing.T) {
  448. t.Setenv("XAI_GROK_CLI_VERSION", "0.2.120-beta.1")
  449. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/responses", nil)
  450. require.NoError(t, err)
  451. applyGrokCLIProxyHeaders(req)
  452. require.Equal(t, xai.CLIClientVersion, req.Header.Get("x-grok-client-version"))
  453. require.Equal(t, xai.CLIUserAgent(xai.CLIClientVersion), req.Header.Get("User-Agent"))
  454. })
  455. // Every entry sits above the pinned minimum, so a rejection here can only be
  456. // caused by the malformed semver and never by the version being too old.
  457. for _, version := range []string{
  458. "0.2.0121",
  459. "0.2.121-alpha..1",
  460. "0.3",
  461. "1",
  462. "0.2.121+build.1",
  463. } {
  464. t.Run("rejects invalid semver "+version, func(t *testing.T) {
  465. t.Setenv("XAI_GROK_CLI_VERSION", version)
  466. req, err := http.NewRequest(http.MethodPost, "https://cli-chat-proxy.grok.com/v1/responses", nil)
  467. require.NoError(t, err)
  468. applyGrokCLIProxyHeaders(req)
  469. require.Equal(t, xai.CLIClientVersion, req.Header.Get("x-grok-client-version"))
  470. require.Equal(t, xai.CLIUserAgent(xai.CLIClientVersion), req.Header.Get("User-Agent"))
  471. })
  472. }
  473. t.Run("leaves direct xAI API requests unchanged", func(t *testing.T) {
  474. t.Setenv("XAI_GROK_CLI_VERSION", "0.2.95")
  475. req, err := http.NewRequest(http.MethodPost, "https://api.x.ai/v1/responses", nil)
  476. require.NoError(t, err)
  477. req.Header.Set("User-Agent", "direct-api-client/1.0")
  478. applyGrokCLIProxyHeaders(req)
  479. require.Empty(t, req.Header.Get("x-grok-client-version"))
  480. require.Empty(t, req.Header.Get("X-XAI-Token-Auth"))
  481. require.Equal(t, "direct-api-client/1.0", req.Header.Get("User-Agent"))
  482. })
  483. }
  484. // HTTPUpstreamSuite HTTP 上游服务测试套件
  485. // 使用 testify/suite 组织测试,支持 SetupTest 初始化
  486. type HTTPUpstreamSuite struct {
  487. suite.Suite
  488. cfg *config.Config // 测试用配置
  489. }
  490. // SetupTest 每个测试用例执行前的初始化
  491. // 创建空配置,各测试用例可按需覆盖
  492. func (s *HTTPUpstreamSuite) SetupTest() {
  493. s.cfg = &config.Config{
  494. Security: config.SecurityConfig{
  495. URLAllowlist: config.URLAllowlistConfig{
  496. AllowPrivateHosts: true,
  497. },
  498. },
  499. }
  500. }
  501. // newService 创建测试用的 httpUpstreamService 实例
  502. // 返回具体类型以便访问内部状态进行断言
  503. func (s *HTTPUpstreamSuite) newService() *httpUpstreamService {
  504. up := NewHTTPUpstream(s.cfg)
  505. svc, ok := up.(*httpUpstreamService)
  506. require.True(s.T(), ok, "expected *httpUpstreamService")
  507. return svc
  508. }
  509. // TestDefaultResponseHeaderTimeout 测试默认响应头超时配置
  510. // 验证显式 0 会禁用等待响应头超时
  511. func (s *HTTPUpstreamSuite) TestDefaultResponseHeaderTimeout() {
  512. svc := s.newService()
  513. entry := mustGetOrCreateClient(s.T(), svc, "", 0, 0)
  514. transport, ok := entry.client.Transport.(*http.Transport)
  515. require.True(s.T(), ok, "expected *http.Transport")
  516. require.Equal(s.T(), time.Duration(0), transport.ResponseHeaderTimeout, "ResponseHeaderTimeout mismatch")
  517. }
  518. // TestNilConfigResponseHeaderTimeoutFallback 验证 nil 配置使用代码级兜底值。
  519. func (s *HTTPUpstreamSuite) TestNilConfigResponseHeaderTimeoutFallback() {
  520. up := NewHTTPUpstream(nil)
  521. svc, ok := up.(*httpUpstreamService)
  522. require.True(s.T(), ok, "expected *httpUpstreamService")
  523. entry := mustGetOrCreateClient(s.T(), svc, "", 0, 0)
  524. transport, ok := entry.client.Transport.(*http.Transport)
  525. require.True(s.T(), ok, "expected *http.Transport")
  526. require.Equal(s.T(), 300*time.Second, transport.ResponseHeaderTimeout, "ResponseHeaderTimeout mismatch")
  527. }
  528. // TestCustomResponseHeaderTimeout 测试自定义响应头超时配置
  529. // 验证配置值能正确应用到 Transport
  530. func (s *HTTPUpstreamSuite) TestCustomResponseHeaderTimeout() {
  531. s.cfg.Gateway = config.GatewayConfig{ResponseHeaderTimeout: 7}
  532. svc := s.newService()
  533. entry := mustGetOrCreateClient(s.T(), svc, "", 0, 0)
  534. transport, ok := entry.client.Transport.(*http.Transport)
  535. require.True(s.T(), ok, "expected *http.Transport")
  536. require.Equal(s.T(), 7*time.Second, transport.ResponseHeaderTimeout, "ResponseHeaderTimeout mismatch")
  537. }
  538. // TestGetOrCreateClient_InvalidURLReturnsError 测试无效代理 URL 返回错误
  539. // 验证解析失败时拒绝回退到直连模式
  540. func (s *HTTPUpstreamSuite) TestGetOrCreateClient_InvalidURLReturnsError() {
  541. svc := s.newService()
  542. _, err := svc.getClientEntry("://bad-proxy-url", 1, 1, service.HTTPUpstreamProfileDefault, false, false)
  543. require.Error(s.T(), err, "expected error for invalid proxy URL")
  544. }
  545. func (s *HTTPUpstreamSuite) TestOpenAIProfileDefaultsToHTTP2AndNoHeaderTimeout() {
  546. s.cfg.Gateway = config.GatewayConfig{
  547. ResponseHeaderTimeout: 600,
  548. OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{
  549. Enabled: true,
  550. AllowProxyFallbackToHTTP1: true,
  551. },
  552. }
  553. svc := s.newService()
  554. entry, err := svc.getClientEntry("", 1, 1, service.HTTPUpstreamProfileOpenAI, false, false)
  555. require.NoError(s.T(), err)
  556. transport, ok := entry.client.Transport.(*http.Transport)
  557. require.True(s.T(), ok, "expected *http.Transport")
  558. require.Equal(s.T(), time.Duration(0), transport.ResponseHeaderTimeout, "OpenAI profile should not inherit generic header timeout")
  559. require.True(s.T(), transport.ForceAttemptHTTP2, "OpenAI profile should prefer HTTP/2")
  560. require.Equal(s.T(), upstreamProtocolModeOpenAIH2, entry.protocolMode)
  561. }
  562. func (s *HTTPUpstreamSuite) TestLongStreamProfileUsesSharedHTTP2KeepAlive() {
  563. s.cfg.Gateway = config.GatewayConfig{
  564. ResponseHeaderTimeout: 600,
  565. OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{
  566. Enabled: false,
  567. },
  568. }
  569. svc := s.newService()
  570. entry, err := svc.getClientEntry("", 1, 1, service.HTTPUpstreamProfileLongStream, false, false)
  571. require.NoError(s.T(), err)
  572. transport, ok := entry.client.Transport.(*http.Transport)
  573. require.True(s.T(), ok, "expected *http.Transport")
  574. require.Equal(s.T(), 600*time.Second, transport.ResponseHeaderTimeout, "long-stream profile should retain the generic header timeout")
  575. require.True(s.T(), transport.ForceAttemptHTTP2, "long-stream profile must enable HTTP/2 independently of OpenAI settings")
  576. require.True(s.T(), transport.Protocols.HTTP2(), "long-stream profile must install HTTP/2 PING health checks")
  577. require.Equal(s.T(), upstreamProtocolModeLongStreamH2, entry.protocolMode)
  578. }
  579. func (s *HTTPUpstreamSuite) TestOpenAIProfileCustomHeaderTimeout() {
  580. s.cfg.Gateway = config.GatewayConfig{
  581. ResponseHeaderTimeout: 600,
  582. OpenAIResponseHeaderTimeout: 1800,
  583. OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{
  584. Enabled: true,
  585. },
  586. }
  587. svc := s.newService()
  588. entry, err := svc.getClientEntry("", 1, 1, service.HTTPUpstreamProfileOpenAI, false, false)
  589. require.NoError(s.T(), err)
  590. transport, ok := entry.client.Transport.(*http.Transport)
  591. require.True(s.T(), ok, "expected *http.Transport")
  592. require.Equal(s.T(), 1800*time.Second, transport.ResponseHeaderTimeout)
  593. }
  594. func (s *HTTPUpstreamSuite) TestOpenAIProfileTLSFingerprintDoesNotInheritGenericHeaderTimeout() {
  595. s.cfg.Gateway = config.GatewayConfig{
  596. ResponseHeaderTimeout: 600,
  597. OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{
  598. Enabled: true,
  599. },
  600. }
  601. svc := s.newService()
  602. entry, err := svc.getClientEntryWithTLS("", 1, 1, &tlsfingerprint.Profile{Name: "test"}, service.HTTPUpstreamProfileOpenAI, false, false)
  603. require.NoError(s.T(), err)
  604. transport, ok := entry.client.Transport.(*http.Transport)
  605. require.True(s.T(), ok, "expected *http.Transport")
  606. require.Equal(s.T(), time.Duration(0), transport.ResponseHeaderTimeout, "OpenAI TLS path should not inherit generic header timeout")
  607. }
  608. func (s *HTTPUpstreamSuite) TestOpenAIProfileHTTP2DisabledUsesHTTP1Transport() {
  609. s.cfg.Gateway = config.GatewayConfig{
  610. OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{Enabled: false},
  611. }
  612. svc := s.newService()
  613. entry, err := svc.getClientEntry("", 1, 1, service.HTTPUpstreamProfileOpenAI, false, false)
  614. require.NoError(s.T(), err)
  615. transport, ok := entry.client.Transport.(*http.Transport)
  616. require.True(s.T(), ok, "expected *http.Transport")
  617. require.False(s.T(), transport.ForceAttemptHTTP2, "OpenAI HTTP/2 disabled should not force H2")
  618. require.NotNil(s.T(), transport.TLSNextProto, "HTTP/1 mode should disable automatic H2 negotiation")
  619. require.Equal(s.T(), upstreamProtocolModeOpenAIH1, entry.protocolMode)
  620. }
  621. func (s *HTTPUpstreamSuite) TestOpenAIHeaderTimeoutChangeRebuildsClient() {
  622. s.cfg.Gateway = config.GatewayConfig{
  623. OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{Enabled: true},
  624. }
  625. svc := s.newService()
  626. entry1, err := svc.getClientEntry("", 1, 1, service.HTTPUpstreamProfileOpenAI, false, false)
  627. require.NoError(s.T(), err)
  628. s.cfg.Gateway.OpenAIResponseHeaderTimeout = 1800
  629. entry2, err := svc.getClientEntry("", 1, 1, service.HTTPUpstreamProfileOpenAI, false, false)
  630. require.NoError(s.T(), err)
  631. require.NotSame(s.T(), entry1, entry2, "OpenAI header timeout changes must rebuild cached client")
  632. transport, ok := entry2.client.Transport.(*http.Transport)
  633. require.True(s.T(), ok, "expected *http.Transport")
  634. require.Equal(s.T(), 1800*time.Second, transport.ResponseHeaderTimeout)
  635. }
  636. func (s *HTTPUpstreamSuite) TestOpenAIHTTP2TimeoutDoesNotActivateProxyFallback() {
  637. s.cfg.Gateway = config.GatewayConfig{
  638. OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{
  639. Enabled: true,
  640. AllowProxyFallbackToHTTP1: true,
  641. FallbackErrorThreshold: 1,
  642. FallbackWindowSeconds: 60,
  643. FallbackTTLSeconds: 600,
  644. },
  645. }
  646. svc := s.newService()
  647. proxyURL := "http://proxy.local:8080"
  648. svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, errors.New("http2: timeout awaiting response headers"))
  649. require.False(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL), "header timeout should not be treated as H2 compatibility failure")
  650. }
  651. func (s *HTTPUpstreamSuite) TestOpenAIHTTP2ProxyCompatibilityErrorActivatesFallback() {
  652. s.cfg.Gateway = config.GatewayConfig{
  653. OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{
  654. Enabled: true,
  655. AllowProxyFallbackToHTTP1: true,
  656. FallbackErrorThreshold: 1,
  657. FallbackWindowSeconds: 60,
  658. FallbackTTLSeconds: 600,
  659. },
  660. }
  661. svc := s.newService()
  662. proxyURL := "http://proxy.local:8080"
  663. svc.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyURL, errors.New("http2: protocol error"))
  664. require.True(s.T(), svc.isOpenAIHTTP2FallbackActive(proxyURL))
  665. entry, err := svc.getClientEntry(proxyURL, 1, 1, service.HTTPUpstreamProfileOpenAI, false, false)
  666. require.NoError(s.T(), err)
  667. transport, ok := entry.client.Transport.(*http.Transport)
  668. require.True(s.T(), ok, "expected *http.Transport")
  669. require.False(s.T(), transport.ForceAttemptHTTP2)
  670. require.NotNil(s.T(), transport.TLSNextProto)
  671. require.Equal(s.T(), upstreamProtocolModeOpenAIH1Fallback, entry.protocolMode)
  672. }
  673. // TestNormalizeProxyURL_Canonicalizes 测试代理 URL 规范化
  674. // 验证等价地址能够映射到同一缓存键
  675. func (s *HTTPUpstreamSuite) TestNormalizeProxyURL_Canonicalizes() {
  676. key1, _, err1 := normalizeProxyURL("http://proxy.local:8080")
  677. require.NoError(s.T(), err1)
  678. key2, _, err2 := normalizeProxyURL("http://proxy.local:8080/")
  679. require.NoError(s.T(), err2)
  680. require.Equal(s.T(), key1, key2, "expected normalized proxy keys to match")
  681. }
  682. // TestAcquireClient_OverLimitReturnsError 测试连接池缓存上限保护
  683. // 验证超限且无可淘汰条目时返回错误
  684. func (s *HTTPUpstreamSuite) TestAcquireClient_OverLimitReturnsError() {
  685. s.cfg.Gateway = config.GatewayConfig{
  686. ConnectionPoolIsolation: config.ConnectionPoolIsolationAccountProxy,
  687. MaxUpstreamClients: 1,
  688. }
  689. svc := s.newService()
  690. entry1, err := svc.acquireClient("http://proxy-a:8080", 1, 1)
  691. require.NoError(s.T(), err, "expected first acquire to succeed")
  692. require.NotNil(s.T(), entry1, "expected entry")
  693. entry2, err := svc.acquireClient("http://proxy-b:8080", 2, 1)
  694. require.Error(s.T(), err, "expected error when cache limit reached")
  695. require.Nil(s.T(), entry2, "expected nil entry when cache limit reached")
  696. }
  697. // TestDo_WithoutProxy_GoesDirect 测试无代理时直连
  698. // 验证空代理 URL 时请求直接发送到目标服务器
  699. func (s *HTTPUpstreamSuite) TestDo_WithoutProxy_GoesDirect() {
  700. // 创建模拟上游服务器
  701. upstream := newLocalTestServer(s.T(), http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  702. _, _ = io.WriteString(w, "direct")
  703. }))
  704. s.T().Cleanup(upstream.Close)
  705. up := NewHTTPUpstream(s.cfg)
  706. req, err := http.NewRequest(http.MethodGet, upstream.URL+"/x", nil)
  707. require.NoError(s.T(), err, "NewRequest")
  708. resp, err := up.Do(req, "", 1, 1)
  709. require.NoError(s.T(), err, "Do")
  710. defer func() { _ = resp.Body.Close() }()
  711. b, _ := io.ReadAll(resp.Body)
  712. require.Equal(s.T(), "direct", string(b), "unexpected body")
  713. }
  714. // TestDo_WithHTTPProxy_UsesProxy 测试 HTTP 代理功能
  715. // 验证请求通过代理服务器转发,使用绝对 URI 格式
  716. func (s *HTTPUpstreamSuite) TestDo_WithHTTPProxy_UsesProxy() {
  717. // 用于接收代理请求的通道
  718. seen := make(chan string, 1)
  719. // 创建模拟代理服务器
  720. proxySrv := newLocalTestServer(s.T(), http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  721. seen <- r.RequestURI // 记录请求 URI
  722. _, _ = io.WriteString(w, "proxied")
  723. }))
  724. s.T().Cleanup(proxySrv.Close)
  725. s.cfg.Gateway = config.GatewayConfig{ResponseHeaderTimeout: 1}
  726. up := NewHTTPUpstream(s.cfg)
  727. // 发送请求到外部地址,应通过代理
  728. req, err := http.NewRequest(http.MethodGet, "http://example.com/test", nil)
  729. require.NoError(s.T(), err, "NewRequest")
  730. resp, err := up.Do(req, proxySrv.URL, 1, 1)
  731. require.NoError(s.T(), err, "Do")
  732. defer func() { _ = resp.Body.Close() }()
  733. b, _ := io.ReadAll(resp.Body)
  734. require.Equal(s.T(), "proxied", string(b), "unexpected body")
  735. // 验证代理收到的是绝对 URI 格式(HTTP 代理规范要求)
  736. select {
  737. case uri := <-seen:
  738. require.Equal(s.T(), "http://example.com/test", uri, "expected absolute-form request URI")
  739. default:
  740. require.Fail(s.T(), "expected proxy to receive request")
  741. }
  742. }
  743. // TestDo_EmptyProxy_UsesDirect 测试空代理字符串
  744. // 验证空字符串代理等同于直连
  745. func (s *HTTPUpstreamSuite) TestDo_EmptyProxy_UsesDirect() {
  746. upstream := newLocalTestServer(s.T(), http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  747. _, _ = io.WriteString(w, "direct-empty")
  748. }))
  749. s.T().Cleanup(upstream.Close)
  750. up := NewHTTPUpstream(s.cfg)
  751. req, err := http.NewRequest(http.MethodGet, upstream.URL+"/y", nil)
  752. require.NoError(s.T(), err, "NewRequest")
  753. resp, err := up.Do(req, "", 1, 1)
  754. require.NoError(s.T(), err, "Do with empty proxy")
  755. defer func() { _ = resp.Body.Close() }()
  756. b, _ := io.ReadAll(resp.Body)
  757. require.Equal(s.T(), "direct-empty", string(b))
  758. }
  759. // TestAccountIsolation_DifferentAccounts 测试账户隔离模式
  760. // 验证不同账户使用独立的连接池
  761. func (s *HTTPUpstreamSuite) TestAccountIsolation_DifferentAccounts() {
  762. s.cfg.Gateway = config.GatewayConfig{ConnectionPoolIsolation: config.ConnectionPoolIsolationAccount}
  763. svc := s.newService()
  764. // 同一代理,不同账户
  765. entry1 := mustGetOrCreateClient(s.T(), svc, "http://proxy.local:8080", 1, 3)
  766. entry2 := mustGetOrCreateClient(s.T(), svc, "http://proxy.local:8080", 2, 3)
  767. require.NotSame(s.T(), entry1, entry2, "不同账号不应共享连接池")
  768. require.Equal(s.T(), 2, len(svc.clients), "账号隔离应缓存两个客户端")
  769. }
  770. // TestAccountProxyIsolation_DifferentProxy 测试账户+代理组合隔离模式
  771. // 验证同一账户使用不同代理时创建独立连接池
  772. func (s *HTTPUpstreamSuite) TestAccountProxyIsolation_DifferentProxy() {
  773. s.cfg.Gateway = config.GatewayConfig{ConnectionPoolIsolation: config.ConnectionPoolIsolationAccountProxy}
  774. svc := s.newService()
  775. // 同一账户,不同代理
  776. entry1 := mustGetOrCreateClient(s.T(), svc, "http://proxy-a:8080", 1, 3)
  777. entry2 := mustGetOrCreateClient(s.T(), svc, "http://proxy-b:8080", 1, 3)
  778. require.NotSame(s.T(), entry1, entry2, "账号+代理隔离应区分不同代理")
  779. require.Equal(s.T(), 2, len(svc.clients), "账号+代理隔离应缓存两个客户端")
  780. }
  781. // TestAccountModeProxyChangeClearsPool 测试账户模式下代理变更
  782. // 验证账户切换代理时清理旧连接池,避免复用错误代理
  783. func (s *HTTPUpstreamSuite) TestAccountModeProxyChangeClearsPool() {
  784. s.cfg.Gateway = config.GatewayConfig{ConnectionPoolIsolation: config.ConnectionPoolIsolationAccount}
  785. svc := s.newService()
  786. // 同一账户,先后使用不同代理
  787. entry1 := mustGetOrCreateClient(s.T(), svc, "http://proxy-a:8080", 1, 3)
  788. entry2 := mustGetOrCreateClient(s.T(), svc, "http://proxy-b:8080", 1, 3)
  789. require.NotSame(s.T(), entry1, entry2, "账号切换代理应创建新连接池")
  790. require.Equal(s.T(), 1, len(svc.clients), "账号模式下应仅保留一个连接池")
  791. require.False(s.T(), hasEntry(svc, entry1), "旧连接池应被清理")
  792. }
  793. // TestAccountConcurrencyOverridesPoolSettings 测试账户并发数覆盖连接池配置
  794. // 验证账户隔离模式下,连接池大小与账户并发数对应
  795. func (s *HTTPUpstreamSuite) TestAccountConcurrencyOverridesPoolSettings() {
  796. s.cfg.Gateway = config.GatewayConfig{ConnectionPoolIsolation: config.ConnectionPoolIsolationAccount}
  797. svc := s.newService()
  798. // 账户并发数为 12
  799. entry := mustGetOrCreateClient(s.T(), svc, "", 1, 12)
  800. transport, ok := entry.client.Transport.(*http.Transport)
  801. require.True(s.T(), ok, "expected *http.Transport")
  802. // 连接池参数应与并发数一致
  803. require.Equal(s.T(), 12, transport.MaxConnsPerHost, "MaxConnsPerHost mismatch")
  804. require.Equal(s.T(), 12, transport.MaxIdleConns, "MaxIdleConns mismatch")
  805. require.Equal(s.T(), 12, transport.MaxIdleConnsPerHost, "MaxIdleConnsPerHost mismatch")
  806. }
  807. // TestAccountConcurrencyFallbackToDefault 测试账户并发数为 0 时回退到默认配置
  808. // 验证未指定并发数时使用全局配置值
  809. func (s *HTTPUpstreamSuite) TestAccountConcurrencyFallbackToDefault() {
  810. s.cfg.Gateway = config.GatewayConfig{
  811. ConnectionPoolIsolation: config.ConnectionPoolIsolationAccount,
  812. MaxIdleConns: 77,
  813. MaxIdleConnsPerHost: 55,
  814. MaxConnsPerHost: 66,
  815. }
  816. svc := s.newService()
  817. // 账户并发数为 0,应使用全局配置
  818. entry := mustGetOrCreateClient(s.T(), svc, "", 1, 0)
  819. transport, ok := entry.client.Transport.(*http.Transport)
  820. require.True(s.T(), ok, "expected *http.Transport")
  821. require.Equal(s.T(), 66, transport.MaxConnsPerHost, "MaxConnsPerHost fallback mismatch")
  822. require.Equal(s.T(), 77, transport.MaxIdleConns, "MaxIdleConns fallback mismatch")
  823. require.Equal(s.T(), 55, transport.MaxIdleConnsPerHost, "MaxIdleConnsPerHost fallback mismatch")
  824. }
  825. // TestEvictOverLimitRemovesOldestIdle 测试超出数量限制时的 LRU 淘汰
  826. // 验证优先淘汰最久未使用的空闲客户端
  827. func (s *HTTPUpstreamSuite) TestEvictOverLimitRemovesOldestIdle() {
  828. s.cfg.Gateway = config.GatewayConfig{
  829. ConnectionPoolIsolation: config.ConnectionPoolIsolationAccountProxy,
  830. MaxUpstreamClients: 2, // 最多缓存 2 个客户端
  831. }
  832. svc := s.newService()
  833. // 创建两个客户端,设置不同的最后使用时间
  834. entry1 := mustGetOrCreateClient(s.T(), svc, "http://proxy-a:8080", 1, 1)
  835. entry2 := mustGetOrCreateClient(s.T(), svc, "http://proxy-b:8080", 2, 1)
  836. atomic.StoreInt64(&entry1.lastUsed, time.Now().Add(-2*time.Hour).UnixNano()) // 最久
  837. atomic.StoreInt64(&entry2.lastUsed, time.Now().Add(-time.Hour).UnixNano())
  838. // 创建第三个客户端,触发淘汰
  839. _ = mustGetOrCreateClient(s.T(), svc, "http://proxy-c:8080", 3, 1)
  840. require.LessOrEqual(s.T(), len(svc.clients), 2, "应保持在缓存上限内")
  841. require.False(s.T(), hasEntry(svc, entry1), "最久未使用的连接池应被清理")
  842. }
  843. // TestIdleTTLDoesNotEvictActive 测试活跃请求保护
  844. // 验证有进行中请求的客户端不会被空闲超时淘汰
  845. func (s *HTTPUpstreamSuite) TestIdleTTLDoesNotEvictActive() {
  846. s.cfg.Gateway = config.GatewayConfig{
  847. ConnectionPoolIsolation: config.ConnectionPoolIsolationAccount,
  848. ClientIdleTTLSeconds: 1, // 1 秒空闲超时
  849. }
  850. svc := s.newService()
  851. entry1 := mustGetOrCreateClient(s.T(), svc, "", 1, 1)
  852. // 设置为很久之前使用,但有活跃请求
  853. atomic.StoreInt64(&entry1.lastUsed, time.Now().Add(-2*time.Minute).UnixNano())
  854. atomic.StoreInt64(&entry1.inFlight, 1) // 模拟有活跃请求
  855. // 创建新客户端,触发淘汰检查
  856. _, _ = svc.getOrCreateClient("", 2, 1)
  857. require.True(s.T(), hasEntry(svc, entry1), "有活跃请求时不应回收")
  858. }
  859. // TestHTTPUpstreamSuite 运行测试套件
  860. func TestHTTPUpstreamSuite(t *testing.T) {
  861. suite.Run(t, new(HTTPUpstreamSuite))
  862. }
  863. // mustGetOrCreateClient 测试辅助函数,调用 getOrCreateClient 并断言无错误
  864. func mustGetOrCreateClient(t *testing.T, svc *httpUpstreamService, proxyURL string, accountID int64, concurrency int) *upstreamClientEntry {
  865. t.Helper()
  866. entry, err := svc.getOrCreateClient(proxyURL, accountID, concurrency)
  867. require.NoError(t, err, "getOrCreateClient(%q, %d, %d)", proxyURL, accountID, concurrency)
  868. return entry
  869. }
  870. // hasEntry 检查客户端是否存在于缓存中
  871. // 辅助函数,用于验证淘汰逻辑
  872. func hasEntry(svc *httpUpstreamService, target *upstreamClientEntry) bool {
  873. for _, entry := range svc.clients {
  874. if entry == target {
  875. return true
  876. }
  877. }
  878. return false
  879. }
  880. func TestHTTPUpstreamDoPublicHostsOnlyRejectsPrivateDestinationBeforeConnecting(t *testing.T) {
  881. var calls atomic.Int64
  882. target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
  883. calls.Add(1)
  884. w.WriteHeader(http.StatusOK)
  885. }))
  886. t.Cleanup(target.Close)
  887. upstream := NewHTTPUpstream(nil)
  888. plain, err := http.NewRequestWithContext(t.Context(), http.MethodGet, target.URL, nil)
  889. require.NoError(t, err)
  890. resp, err := upstream.Do(plain, "", 1, 1)
  891. require.NoError(t, err)
  892. require.NoError(t, resp.Body.Close())
  893. require.Equal(t, int64(1), calls.Load(), "loopback stays reachable for requests without the marker")
  894. guarded, err := http.NewRequestWithContext(service.WithHTTPUpstreamPublicHostsOnly(t.Context()), http.MethodGet, target.URL, nil)
  895. require.NoError(t, err)
  896. resp, err = upstream.Do(guarded, "", 1, 1)
  897. require.Error(t, err)
  898. require.Nil(t, resp)
  899. require.Contains(t, err.Error(), "not allowed")
  900. require.Equal(t, int64(1), calls.Load(), "marked request must be rejected before any connection is made")
  901. }
  902. func TestHTTPUpstreamPublicHostsOnlyValidatesEveryRedirectHop(t *testing.T) {
  903. upstream, ok := NewHTTPUpstream(nil).(*httpUpstreamService)
  904. require.True(t, ok)
  905. base := &http.Client{}
  906. plain, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "https://cdn.example.com/a.png", nil)
  907. require.NoError(t, err)
  908. require.Same(t, base, upstream.httpClientForUpstreamRequest(base, plain))
  909. guarded, err := http.NewRequestWithContext(service.WithHTTPUpstreamPublicHostsOnly(t.Context()), http.MethodGet, "https://cdn.example.com/a.png", nil)
  910. require.NoError(t, err)
  911. client := upstream.httpClientForUpstreamRequest(base, guarded)
  912. require.NotSame(t, base, client)
  913. require.NotNil(t, client.CheckRedirect)
  914. require.Nil(t, base.CheckRedirect, "the cached client must stay untouched")
  915. via := []*http.Request{guarded}
  916. for _, hop := range []string{
  917. "http://127.0.0.1:8080/a.png",
  918. "http://[::1]:8080/a.png",
  919. "http://10.0.0.8/a.png",
  920. "http://169.254.169.254/latest/meta-data/",
  921. "http://0.0.0.0/a.png",
  922. } {
  923. hopReq, err := http.NewRequestWithContext(guarded.Context(), http.MethodGet, hop, nil)
  924. require.NoError(t, err)
  925. require.Error(t, client.CheckRedirect(hopReq, via), "hop=%s", hop)
  926. }
  927. publicHop, err := http.NewRequestWithContext(guarded.Context(), http.MethodGet, "http://93.184.216.34/a.png", nil)
  928. require.NoError(t, err)
  929. require.NoError(t, client.CheckRedirect(publicHop, via))
  930. require.Error(t, client.CheckRedirect(publicHop, make([]*http.Request, 10)), "redirect chain stays capped")
  931. }