CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-08-28 12:44:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

🪢 fix: Harden Langfuse Media Upload Targets (#14974)

Browse source
This commit is contained in:
Ravi Kumar L 2026-08-19 04:06:33 +02:00 • committed by GitHub
parent e4d6bb71f9
commit 5f95631283
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 98 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

26
otel/langfuse-fanout/cmd/langfuse-fanout/main.go
View file

@ -658,6 +658,9 @@ func (g *gateway) patchMedia(ctx context.Context, dest destination, path string,
}
func (g *gateway) putMedia(ctx context.Context, dest uploadDestination, body []byte, originalHeaders http.Header) (int, error) {
if err := validateMediaUploadURL(dest.UploadURL); err != nil {
return 0, err
}
req, err := http.NewRequestWithContext(ctx, http.MethodPut, dest.UploadURL, bytes.NewReader(body))
if err != nil {
return 0, err
@ -680,7 +683,11 @@ func (g *gateway) putMedia(ctx context.Context, dest uploadDestination, body []b
req.Header.Set("x-amz-checksum-sha256", value)
}
}
resp, err := g.doUpstream(req, "media_upload", dest.Name)
uploadClient := *g.cfg.client
uploadClient.CheckRedirect = func(_ *http.Request, _ []*http.Request) error {
return http.ErrUseLastResponse
}
resp, err := g.doUpstreamWithClient(&uploadClient, req, "media_upload", dest.Name)
if err != nil {
return 0, err
}
@ -706,8 +713,12 @@ func (g *gateway) doExpect2xx(operation string, destination string, req *http.Re
}
func (g *gateway) doUpstream(req *http.Request, operation string, destination string) (*http.Response, error) {
return g.doUpstreamWithClient(g.cfg.client, req, operation, destination)
}
func (g *gateway) doUpstreamWithClient(client *http.Client, req *http.Request, operation string, destination string) (*http.Response, error) {
startedAt := time.Now()
resp, err := g.cfg.client.Do(req)
resp, err := client.Do(req)
if err != nil {
duration := time.Since(startedAt)
if g.metrics != nil {
@ -1141,6 +1152,17 @@ func isGCSUploadURL(value string) bool {
return host == "storage.googleapis.com" || strings.HasSuffix(host, ".storage.googleapis.com")
}
func validateMediaUploadURL(value string) error {
parsed, err := url.Parse(value)
if err != nil || parsed.Hostname() == "" {
return errors.New("media upload URL must be an absolute HTTPS URL")
}
if parsed.Scheme != "https" {
return errors.New("media upload URL must use HTTPS")
}
return nil
}
func isAzureUploadURL(value string) bool {
parsed, err := url.Parse(value)
if err != nil {

81
otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go
View file

@ -99,6 +99,66 @@ func TestNormalizeBaseURLAllowsOnlyHTTPAndHTTPS(t *testing.T) {
}
}
func TestValidateMediaUploadURLRequiresHTTPS(t *testing.T) {
t.Parallel()
tests := []struct {
name string
url string
wantErr bool
}{
{name: "public storage", url: "https://bucket.s3.amazonaws.com/object?X-Amz-Signature=value"},
{name: "self-hosted storage", url: "https://minio.internal:9000/object"},
{name: "private address", url: "https://10.0.0.8/object"},
{name: "http", url: "http://minio.internal:9000/object", wantErr: true},
{name: "unsupported scheme", url: "ftp://storage.example.com/object", wantErr: true},
{name: "relative", url: "/object", wantErr: true},
{name: "missing host", url: "https:///object", wantErr: true},
{name: "malformed", url: "://storage.example.com/object", wantErr: true},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
err := validateMediaUploadURL(test.url)
if (err != nil) != test.wantErr {
t.Fatalf("validateMediaUploadURL(%q) error = %v, wantErr %t", test.url, err, test.wantErr)
}
})
}
}
func TestPutMediaDoesNotFollowRedirects(t *testing.T) {
t.Parallel()
var targetRequests int
target := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
targetRequests++
w.WriteHeader(http.StatusOK)
}))
defer target.Close()
redirect := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, target.URL+"/upload", http.StatusTemporaryRedirect)
}))
defer redirect.Close()
gw := newTestGateway(redirect.URL, nil)
gw.cfg.client = redirect.Client()
status, err := gw.putMedia(context.Background(), uploadDestination{
Name: centralName,
UploadURL: redirect.URL + "/upload",
}, []byte("hello"), http.Header{"Content-Type": []string{"image/png"}})
if status != http.StatusTemporaryRedirect {
t.Fatalf("status = %d, want %d", status, http.StatusTemporaryRedirect)
}
if err == nil {
t.Fatal("expected redirect response to fail the upload")
}
if targetRequests != 0 {
t.Fatalf("redirect target requests = %d, want 0", targetRequests)
}
}
func TestTraceProxyForwardsExistingRoutingAttributesToCollector(t *testing.T) {
t.Parallel()
@ -290,10 +350,10 @@ func TestMediaUploadFansOutToCentralAndTenant(t *testing.T) {
var mu sync.Mutex
uploads := map[string]string{}
upstream := func(name string) *httptest.Server {
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
return httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.Method == http.MethodPost && r.URL.Path == mediaPath:
uploadURL := "http://" + r.Host + "/upload/" + name
uploadURL := "https://" + r.Host + "/upload/" + name
writeJSON(w, http.StatusCreated, mediaUploadResponse{
MediaID: "same-media-id",
UploadURL: &uploadURL,
@ -319,6 +379,8 @@ func TestMediaUploadFansOutToCentralAndTenant(t *testing.T) {
store := newFakeUploadPlanStore()
createGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store)
uploadGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store)
createGateway.cfg.client = central.Client()
uploadGateway.cfg.client = central.Client()
createBody := `{"traceId":"trace","contentType":"image/png","contentLength":5,"sha256Hash":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","field":"input"}`
req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+mediaPath, strings.NewReader(createBody))
req.Header.Set("Authorization", "Basic tenant")
@ -376,10 +438,10 @@ func TestMediaUploadSkipsCentralForCentralMediaDisabledTenantRoute(t *testing.T)
var mu sync.Mutex
uploads := map[string]string{}
upstream := func(name string) *httptest.Server {
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
return httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.Method == http.MethodPost && r.URL.Path == mediaPath:
uploadURL := "http://" + r.Host + "/upload/" + name
uploadURL := "https://" + r.Host + "/upload/" + name
writeJSON(w, http.StatusCreated, mediaUploadResponse{
MediaID: "same-media-id",
UploadURL: &uploadURL,
@ -405,6 +467,8 @@ func TestMediaUploadSkipsCentralForCentralMediaDisabledTenantRoute(t *testing.T)
store := newFakeUploadPlanStore()
createGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store)
uploadGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store)
createGateway.cfg.client = central.Client()
uploadGateway.cfg.client = central.Client()
createBody := `{"traceId":"trace","contentType":"image/png","contentLength":5,"sha256Hash":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","field":"input"}`
req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu/"+centralMediaDisabled+mediaPath, strings.NewReader(createBody))
req.Header.Set("Authorization", "Basic tenant")
@ -588,7 +652,7 @@ func TestMediaUploadIsOneTime(t *testing.T) {
t.Parallel()
var uploads int
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPut || r.URL.Path != "/upload" {
http.NotFound(w, r)
return
@ -609,6 +673,7 @@ func TestMediaUploadIsOneTime(t *testing.T) {
}},
})
gw := newTestGatewayWithStore(upstream.URL, nil, store)
gw.cfg.client = upstream.Client()
for index, expectedStatus := range []int{http.StatusOK, http.StatusNotFound} {
req := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("hello"))
@ -628,7 +693,7 @@ func TestMediaUploadOversizeRestoresPlanForRetry(t *testing.T) {
t.Parallel()
var uploads int
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPut || r.URL.Path != "/upload" {
http.NotFound(w, r)
return
@ -649,6 +714,7 @@ func TestMediaUploadOversizeRestoresPlanForRetry(t *testing.T) {
}},
})
gw := newTestGatewayWithStore(upstream.URL, nil, store)
gw.cfg.client = upstream.Client()
oversizeReq := httptest.NewRequest(
http.MethodPut,
@ -678,7 +744,7 @@ func TestMediaUploadUnsupportedContentTypeRestoresPlanForRetry(t *testing.T) {
t.Parallel()
var uploads int
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPut || r.URL.Path != "/upload" {
http.NotFound(w, r)
return
@ -699,6 +765,7 @@ func TestMediaUploadUnsupportedContentTypeRestoresPlanForRetry(t *testing.T) {
}},
})
gw := newTestGatewayWithStore(upstream.URL, nil, store)
gw.cfg.client = upstream.Client()
badReq := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("hello"))
badReq.Header.Set("Content-Type", "text/html")