fix(grok): proxy protected video content through upstream account

(cherry picked from commit a01177e294)
This commit is contained in:
Vz7797
2026-07-18 14:37:25 +08:00
committed by Heatherm Huang
parent b1a6b80267
commit 3f6b5c7bd7
8 changed files with 340 additions and 6 deletions
+8 -3
View File
@@ -46,6 +46,11 @@ func (h *OpenAIGatewayHandler) GrokVideoStatus(c *gin.Context) {
h.handleGrokMedia(c, service.GrokMediaEndpointVideoStatus, c.Param("request_id"))
}
// GrokVideoContent proxies downloadable video content through the task's upstream account.
func (h *OpenAIGatewayHandler) GrokVideoContent(c *gin.Context) {
h.handleGrokMedia(c, service.GrokMediaEndpointVideoContent, c.Param("request_id"))
}
func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service.GrokMediaEndpoint, requestID string) {
streamStarted := false
defer h.recoverResponsesPanic(c, &streamStarted)
@@ -99,7 +104,7 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service.
h.errorResponse(c, http.StatusBadRequest, "invalid_request_error", "model is required")
return
}
if endpoint == service.GrokMediaEndpointVideoStatus && strings.TrimSpace(requestID) == "" {
if endpoint.IsVideoLookupRequest() && strings.TrimSpace(requestID) == "" {
h.errorResponse(c, http.StatusBadRequest, "invalid_request_error", "request_id is required")
return
}
@@ -159,7 +164,7 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service.
sessionSeed = []byte(requestID)
}
sessionHash := h.gatewayService.GenerateExplicitSessionHash(c, sessionSeed)
if endpoint == service.GrokMediaEndpointVideoStatus {
if endpoint.IsVideoLookupRequest() {
sessionHash = service.GrokMediaVideoRequestSessionHash(requestID)
}
requestCtx := c.Request.Context()
@@ -334,7 +339,7 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service.
continue
}
h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, account.GetMappedModel(requestModel), false, nil)
if c.Writer.Size() == writerSizeBeforeForward {
if !service.IsResponseCommitted(c) && c.Writer.Size() == writerSizeBeforeForward {
h.errorResponse(c, http.StatusBadGateway, "upstream_error", "Upstream request failed")
}
reqLog.Warn("grok_media.forward_failed",
@@ -38,6 +38,12 @@ func TestShouldRecordGrokMediaUsage(t *testing.T) {
model: "",
want: false,
},
{
name: "video content skips usage",
endpoint: service.GrokMediaEndpointVideoContent,
model: "",
want: false,
},
{
name: "generation skips usage without model",
endpoint: service.GrokMediaEndpointImagesGenerations,
@@ -65,6 +71,7 @@ func TestGrokMediaRequiredCapability(t *testing.T) {
{name: "video edit", endpoint: service.GrokMediaEndpointVideosEdits, want: service.OpenAIEndpointCapabilityGrokMediaGeneration},
{name: "video extension", endpoint: service.GrokMediaEndpointVideosExtensions, want: service.OpenAIEndpointCapabilityGrokMediaGeneration},
{name: "video status preserves lookup", endpoint: service.GrokMediaEndpointVideoStatus, want: ""},
{name: "video content preserves lookup", endpoint: service.GrokMediaEndpointVideoContent, want: ""},
}
for _, tt := range tests {
+15
View File
@@ -91,6 +91,19 @@ func RegisterGatewayRoutes(
},
})
}
videoContentHandler := func(c *gin.Context) {
if getGroupPlatform(c) == service.PlatformGrok {
h.OpenAIGateway.GrokVideoContent(c)
return
}
service.MarkOpsClientBusinessLimited(c, service.OpsClientBusinessLimitedReasonLocalFeatureGate)
c.JSON(http.StatusNotFound, gin.H{
"error": gin.H{
"type": "not_found_error",
"message": "Videos API is not supported for this platform",
},
})
}
videoEditHandler := func(c *gin.Context) {
if getGroupPlatform(c) == service.PlatformGrok {
h.OpenAIGateway.GrokVideoEdit(c)
@@ -209,6 +222,7 @@ func RegisterGatewayRoutes(
gateway.POST("/videos/edits", videoEditHandler)
gateway.POST("/videos/extensions", videoExtensionHandler)
gateway.GET("/videos/:request_id", videoStatusHandler)
gateway.GET("/videos/:request_id/content", videoContentHandler)
}
// Gemini 原生 API 兼容层(Gemini SDK/CLI 直连)
@@ -282,6 +296,7 @@ func RegisterGatewayRoutes(
r.POST("/videos/edits", bodyLimit, clientRequestID, opsErrorLogger, endpointNorm, gin.HandlerFunc(apiKeyAuth), requireGroupAnthropic, videoEditHandler)
r.POST("/videos/extensions", bodyLimit, clientRequestID, opsErrorLogger, endpointNorm, gin.HandlerFunc(apiKeyAuth), requireGroupAnthropic, videoExtensionHandler)
r.GET("/videos/:request_id", bodyLimit, clientRequestID, opsErrorLogger, endpointNorm, gin.HandlerFunc(apiKeyAuth), requireGroupAnthropic, videoStatusHandler)
r.GET("/videos/:request_id/content", bodyLimit, clientRequestID, opsErrorLogger, endpointNorm, gin.HandlerFunc(apiKeyAuth), requireGroupAnthropic, videoContentHandler)
// Antigravity 模型列表
r.GET("/antigravity/models", gin.HandlerFunc(apiKeyAuth), requireGroupAnthropic, h.Gateway.AntigravityModels)
@@ -160,6 +160,8 @@ func TestGatewayRoutesGrokImagesAndVideosPathsAreRegistered(t *testing.T) {
for _, path := range []string{
"/v1/videos/request-123",
"/videos/request-123",
"/v1/videos/request-123/content",
"/videos/request-123/content",
} {
req := httptest.NewRequest(http.MethodGet, path, nil)
w := httptest.NewRecorder()
@@ -186,6 +188,8 @@ func TestGatewayRoutesNonGrokVideosAreRejectedAtPlatformGate(t *testing.T) {
{http.MethodPost, "/videos/extensions", `{"model":"grok-imagine-video","prompt":"waves","video":{"url":"https://example.com/in.mp4"}}`},
{http.MethodGet, "/v1/videos/request-123", ""},
{http.MethodGet, "/videos/request-123", ""},
{http.MethodGet, "/v1/videos/request-123/content", ""},
{http.MethodGet, "/videos/request-123/content", ""},
} {
req := httptest.NewRequest(tc.method, tc.path, strings.NewReader(tc.body))
req.Header.Set("Content-Type", "application/json")
+175 -3
View File
@@ -29,10 +29,15 @@ const (
GrokMediaEndpointVideosEdits GrokMediaEndpoint = "videos_edits"
GrokMediaEndpointVideosExtensions GrokMediaEndpoint = "videos_extensions"
GrokMediaEndpointVideoStatus GrokMediaEndpoint = "video_status"
GrokMediaEndpointVideoContent GrokMediaEndpoint = "video_content"
)
func (e GrokMediaEndpoint) RequiresRequestBody() bool {
return e != GrokMediaEndpointVideoStatus
return !e.IsVideoLookupRequest()
}
func (e GrokMediaEndpoint) IsVideoLookupRequest() bool {
return e == GrokMediaEndpointVideoStatus || e == GrokMediaEndpointVideoContent
}
func (e GrokMediaEndpoint) IsGenerationRequest() bool {
@@ -97,7 +102,7 @@ func (r GrokMediaRequestInfo) ModerationBody() []byte {
}
func (e GrokMediaEndpoint) httpMethod() string {
if e == GrokMediaEndpointVideoStatus {
if e.IsVideoLookupRequest() {
return http.MethodGet
}
return http.MethodPost
@@ -327,7 +332,16 @@ func (s *OpenAIGatewayService) ForwardGrokMedia(
return nil, err
}
upstreamReq.Header.Set("Authorization", "Bearer "+token)
upstreamReq.Header.Set("Accept", "application/json")
if endpoint == GrokMediaEndpointVideoContent {
upstreamReq.Header.Set("Accept", "*/*")
if c != nil {
if rangeHeader := strings.TrimSpace(c.GetHeader("Range")); rangeHeader != "" {
upstreamReq.Header.Set("Range", rangeHeader)
}
}
} else {
upstreamReq.Header.Set("Accept", "application/json")
}
if account.IsGrokOAuth() && isGrokCLIProxyTarget(targetURL) {
applyGrokCLIHeaders(upstreamReq.Header)
}
@@ -360,10 +374,27 @@ func (s *OpenAIGatewayService) ForwardGrokMedia(
}
s.updateGrokUsageFromResponse(ctx, account, resp.Header, resp.StatusCode)
if endpoint == GrokMediaEndpointVideoContent {
if err := writeGrokMediaContentResponse(c, resp); err != nil {
return nil, err
}
return &OpenAIForwardResult{
RequestID: requestIDHeader,
ResponseHeaders: resp.Header.Clone(),
Duration: time.Since(startTime),
}, nil
}
respBody, err := ReadUpstreamResponseBody(resp.Body, s.cfg, c, openAITooLargeError)
if err != nil {
return nil, err
}
if endpoint == GrokMediaEndpointVideoStatus {
respBody = rewriteGrokMediaVideoContentURLs(
respBody,
requestID,
grokMediaContentProxyURL(c, requestID),
)
}
writeGrokMediaResponse(c, resp, respBody, s.responseHeaderFilter)
usage := grokMediaUsageFromResponse(endpoint, requestInfo, respBody)
return &OpenAIForwardResult{
@@ -739,3 +770,144 @@ func writeGrokMediaResponse(c *gin.Context, resp *http.Response, body []byte, fi
}
c.Data(resp.StatusCode, contentType, body)
}
func writeGrokMediaContentResponse(c *gin.Context, resp *http.Response) error {
if c == nil || resp == nil || resp.Body == nil {
return fmt.Errorf("grok media content response is incomplete")
}
for _, name := range []string{
"Content-Type",
"Content-Length",
"Content-Range",
"Accept-Ranges",
"Content-Disposition",
} {
if value := strings.TrimSpace(resp.Header.Get(name)); value != "" {
c.Header(name, value)
}
}
if strings.TrimSpace(c.Writer.Header().Get("Content-Length")) == "" && resp.ContentLength >= 0 {
c.Header("Content-Length", strconv.FormatInt(resp.ContentLength, 10))
}
if strings.TrimSpace(c.Writer.Header().Get("Content-Type")) == "" {
c.Header("Content-Type", "application/octet-stream")
}
c.Status(resp.StatusCode)
MarkResponseCommitted(c)
_, err := io.Copy(c.Writer, resp.Body)
return err
}
func rewriteGrokMediaVideoContentURLs(body []byte, requestID, proxyURL string) []byte {
if len(body) == 0 || strings.TrimSpace(requestID) == "" || strings.TrimSpace(proxyURL) == "" || !gjson.ValidBytes(body) {
return body
}
decoder := json.NewDecoder(bytes.NewReader(body))
decoder.UseNumber()
var value any
if err := decoder.Decode(&value); err != nil {
return body
}
if !rewriteGrokMediaVideoContentURLValue(&value, requestID, proxyURL) {
return body
}
rewritten, err := json.Marshal(value)
if err != nil {
return body
}
return rewritten
}
func rewriteGrokMediaVideoContentURLValue(value *any, requestID, proxyURL string) bool {
if value == nil {
return false
}
switch typed := (*value).(type) {
case map[string]any:
changed := false
for key, child := range typed {
childValue := child
if rewriteGrokMediaVideoContentURLValue(&childValue, requestID, proxyURL) {
typed[key] = childValue
changed = true
}
}
return changed
case []any:
changed := false
for index, child := range typed {
childValue := child
if rewriteGrokMediaVideoContentURLValue(&childValue, requestID, proxyURL) {
typed[index] = childValue
changed = true
}
}
return changed
case string:
if isGrokMediaVideoContentURL(typed, requestID) {
*value = proxyURL
return true
}
}
return false
}
func isGrokMediaVideoContentURL(rawURL, requestID string) bool {
parsed, err := url.Parse(strings.TrimSpace(rawURL))
if err != nil || parsed.Path == "" {
return false
}
segments := strings.Split(strings.Trim(parsed.Path, "/"), "/")
if len(segments) < 3 {
return false
}
requestID = strings.Trim(requestID, "/")
decodedID, err := url.PathUnescape(segments[len(segments)-2])
if err != nil {
return false
}
return segments[len(segments)-3] == "videos" &&
decodedID == requestID &&
segments[len(segments)-1] == "content"
}
func grokMediaContentProxyURL(c *gin.Context, requestID string) string {
if c == nil || c.Request == nil || c.Request.URL == nil || strings.TrimSpace(requestID) == "" {
return ""
}
pathPrefix := ""
if strings.HasPrefix(c.Request.URL.Path, "/v1/") {
pathPrefix = "/v1"
}
path := pathPrefix + "/videos/" + url.PathEscape(strings.Trim(requestID, "/")) + "/content"
host := strings.TrimSpace(c.GetHeader("X-Forwarded-Host"))
if comma := strings.IndexByte(host, ','); comma >= 0 {
host = strings.TrimSpace(host[:comma])
}
if host == "" {
host = strings.TrimSpace(c.Request.Host)
}
if strings.ContainsAny(host, " \t\r\n") {
return ""
}
scheme := strings.ToLower(strings.TrimSpace(c.GetHeader("X-Forwarded-Proto")))
if comma := strings.IndexByte(scheme, ','); comma >= 0 {
scheme = strings.TrimSpace(scheme[:comma])
}
if scheme != "http" && scheme != "https" {
scheme = strings.ToLower(strings.TrimSpace(c.Request.URL.Scheme))
}
if scheme != "http" && scheme != "https" {
if c.Request.TLS != nil {
scheme = "https"
} else {
scheme = "http"
}
}
if host == "" {
return path
}
return scheme + "://" + host + path
}
@@ -0,0 +1,121 @@
//go:build unit
package service
import (
"context"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
)
type grokMediaContentUpstreamStub struct {
request *http.Request
response *http.Response
}
func (s *grokMediaContentUpstreamStub) Do(req *http.Request, _ string, _ int64, _ int) (*http.Response, error) {
s.request = req
return s.response, nil
}
func (s *grokMediaContentUpstreamStub) DoWithTLS(req *http.Request, proxyURL string, accountID int64, accountConcurrency int, _ *tlsfingerprint.Profile) (*http.Response, error) {
return s.Do(req, proxyURL, accountID, accountConcurrency)
}
func grokMediaContentTestAccount() *Account {
return &Account{
ID: 9,
Platform: PlatformGrok,
Type: AccountTypeAPIKey,
Credentials: map[string]any{
"api_key": "upstream-key",
"base_url": "https://relay.example/v1",
},
}
}
func grokMediaContentTestContext(method, target string, headers map[string]string) (*gin.Context, *httptest.ResponseRecorder) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
c, _ := gin.CreateTestContext(recorder)
c.Request = httptest.NewRequest(method, target, nil)
for name, value := range headers {
c.Request.Header.Set(name, value)
}
return c, recorder
}
func TestForwardGrokMediaContentUsesUpstreamCredentialAndStreamsRange(t *testing.T) {
upstream := &grokMediaContentUpstreamStub{
response: &http.Response{
StatusCode: http.StatusPartialContent,
Header: http.Header{
"Content-Type": []string{"video/mp4"},
"Content-Length": []string{"13"},
"Content-Range": []string{"bytes 0-12/100"},
"Accept-Ranges": []string{"bytes"},
"Content-Disposition": []string{
`attachment; filename="task-1.mp4"`,
},
},
Body: io.NopCloser(strings.NewReader("video-payload")),
},
}
svc := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream}
c, recorder := grokMediaContentTestContext(http.MethodGet, "https://api.example/v1/videos/task-1/content", map[string]string{
"Range": "bytes=0-12",
})
result, err := svc.ForwardGrokMedia(
context.Background(), c, grokMediaContentTestAccount(),
GrokMediaEndpointVideoContent, "task-1", nil, "",
)
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, http.StatusPartialContent, recorder.Code)
require.Equal(t, "video-payload", recorder.Body.String())
require.Equal(t, "https://relay.example/v1/videos/task-1/content", upstream.request.URL.String())
require.Equal(t, "Bearer upstream-key", upstream.request.Header.Get("Authorization"))
require.Equal(t, "bytes=0-12", upstream.request.Header.Get("Range"))
require.Equal(t, "*/*", upstream.request.Header.Get("Accept"))
require.Equal(t, "video/mp4", recorder.Header().Get("Content-Type"))
require.Equal(t, "13", recorder.Header().Get("Content-Length"))
require.Equal(t, "bytes 0-12/100", recorder.Header().Get("Content-Range"))
require.Equal(t, "bytes", recorder.Header().Get("Accept-Ranges"))
require.Equal(t, `attachment; filename="task-1.mp4"`, recorder.Header().Get("Content-Disposition"))
}
func TestForwardGrokVideoStatusRewritesOnlyProtectedContentURL(t *testing.T) {
statusBody := `{"id":"task-1","status":"completed","url":"https://relay.example/v1/videos/task-1/content","download_url":"/v1/videos/task-1/content","video_url":"https://vidgen.x.ai/task-1.mp4","counter":9007199254740993}`
upstream := &grokMediaContentUpstreamStub{
response: &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(statusBody)),
},
}
svc := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream}
c, recorder := grokMediaContentTestContext(http.MethodGet, "https://api.example/v1/videos/task-1", nil)
_, err := svc.ForwardGrokMedia(
context.Background(), c, grokMediaContentTestAccount(),
GrokMediaEndpointVideoStatus, "task-1", nil, "",
)
require.NoError(t, err)
require.Equal(t, http.StatusOK, recorder.Code)
require.Equal(t, "https://api.example/v1/videos/task-1/content", gjson.Get(recorder.Body.String(), "url").String())
require.Equal(t, "https://api.example/v1/videos/task-1/content", gjson.Get(recorder.Body.String(), "download_url").String())
require.Equal(t, "https://vidgen.x.ai/task-1.mp4", gjson.Get(recorder.Body.String(), "video_url").String())
require.Equal(t, "9007199254740993", gjson.Get(recorder.Body.String(), "counter").String())
}
@@ -112,6 +112,12 @@ func buildGrokMediaURL(account *Account, cfg *config.Config, endpoint GrokMediaE
return xai.BuildVideosExtensionsURLWithValidator(baseURL, validator)
case GrokMediaEndpointVideoStatus:
return xai.BuildVideoURLWithValidator(baseURL, requestID, validator)
case GrokMediaEndpointVideoContent:
videoURL, err := xai.BuildVideoURLWithValidator(baseURL, requestID, validator)
if err != nil {
return "", err
}
return videoURL + "/content", nil
default:
return "", fmt.Errorf("unsupported grok media endpoint: %s", endpoint)
}
@@ -35,6 +35,10 @@ func TestGrokAPIKeyURLPolicyFollowsGlobalSecurityConfig(t *testing.T) {
mediaURL, err := buildGrokMediaURL(account, cfg, GrokMediaEndpointImagesGenerations, "")
require.NoError(t, err)
require.Equal(t, "http://grok.example.test/v1/images/generations", mediaURL)
contentURL, err := buildGrokMediaURL(account, cfg, GrokMediaEndpointVideoContent, "request 123")
require.NoError(t, err)
require.Equal(t, "http://grok.example.test/v1/videos/request%20123/content", contentURL)
})
t.Run("insecure HTTP disabled", func(t *testing.T) {