diff --git a/pkg/service/agentendpoint_test.go b/pkg/service/agentendpoint_test.go index b9c59343f..9c1323813 100644 --- a/pkg/service/agentendpoint_test.go +++ b/pkg/service/agentendpoint_test.go @@ -41,7 +41,6 @@ import ( "github.com/stretchr/testify/require" "github.com/livekit/livekit-server/pkg/agent" - "github.com/livekit/livekit-server/pkg/agent/endpoint" "github.com/livekit/livekit-server/pkg/agent/endpoint/conformance" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" @@ -56,6 +55,9 @@ const ( testSecret = "verysecretsecret" ) +// small enough that a test body can exceed it +const testMaxAPIBodySize = 64 << 10 + type endpointStack struct { t *testing.T ts *httptest.Server @@ -88,25 +90,29 @@ func newEndpointStack(t *testing.T, endpointsCfg agent.EndpointsConfig) *endpoin require.NoError(t, err) keyProvider := auth.NewSimpleKeyProvider(testKey, testSecret) - svc, err := service.NewAgentService( - &config.Config{ - Region: "test", - Keys: map[string]string{testKey: testSecret}, - Agents: agent.Config{TargetLoad: agent.DefaultTargetLoad, Endpoints: endpointsCfg}, - }, - localNode, - psrpc.NewLocalMessageBus(), - keyProvider, - ) + conf := &config.Config{ + Region: "test", + Keys: map[string]string{testKey: testSecret}, + Agents: agent.Config{TargetLoad: agent.DefaultTargetLoad, Endpoints: endpointsCfg}, + } + conf.Limit.MaxAPIRequestBodySize = testMaxAPIBodySize + + svc, err := service.NewAgentService(conf, localNode, psrpc.NewLocalMessageBus(), keyProvider) require.NoError(t, err) - // public front (client-facing HTTP/1.1), unchanged - mux := http.NewServeMux() - mux.Handle(endpoint.PathPrefix, svc.EndpointFront()) - authMW := service.NewAPIKeyAuthMiddleware(keyProvider) - ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - authMW.ServeHTTP(w, r, mux.ServeHTTP) - })) + // the production handler, so these tests run on the node's real middleware chain + apiMux := http.NewServeMux() + apiMux.HandleFunc("POST /api-sink", func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + w.WriteHeader(http.StatusNoContent) + }) + apiMux.HandleFunc("/", http.NotFound) + + var agentFront http.Handler + if !endpointsCfg.Disabled { + agentFront = svc.EndpointFront() + } + ts := httptest.NewServer(service.NewHTTPHandler(conf, keyProvider, apiMux, agentFront)) t.Cleanup(ts.Close) t.Cleanup(func() { svc.DrainConnections(time.Millisecond, true) }) diff --git a/pkg/service/agentendpointchain_test.go b/pkg/service/agentendpointchain_test.go new file mode 100644 index 000000000..ec9aaa669 --- /dev/null +++ b/pkg/service/agentendpointchain_test.go @@ -0,0 +1,209 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service_test + +import ( + "bufio" + "bytes" + "fmt" + "io" + "net" + "net/http" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/agent" + "github.com/livekit/protocol/auth" + "github.com/livekit/protocol/livekit" +) + +// rawHTTPTarget speaks HTTP/1.1 by hand, so it can end a response in ways +// net/http will not produce. +func rawHTTPTarget(t *testing.T, respond func(net.Conn)) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = ln.Close() }) + + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + go func() { + defer c.Close() + br := bufio.NewReader(c) + for { + line, err := br.ReadString('\n') + if err != nil { + return + } + if line == "\r\n" { + break + } + } + respond(c) + }() + } + }() + return ln.Addr().String() +} + +// A chunked body cut short must not reach the client as complete, and nothing in +// the serving chain may recover the abort that prevents it. +func TestAgentChainDeliversTruncatedChunkedResponse(t *testing.T) { + addr := rawHTTPTarget(t, func(c net.Conn) { + _, _ = io.WriteString(c, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n") + _, _ = fmt.Fprintf(c, "%x\r\n", 1000) + _, _ = c.Write(bytes.Repeat([]byte("y"), 1000)) + _, _ = io.WriteString(c, "\r\n") + // closes without the terminating 0-length chunk + }) + + stack := newEndpointStack(t, agent.EndpointsConfig{}) + stack.startWorker(addr, "production", []*livekit.AgentHttp_AgentEndpoint{ + httpEP("/partial", []string{"GET"}, true), + }) + + resp, err := http.Get(stack.ts.URL + "/agents/test-agent/production/partial") + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + + body, err := io.ReadAll(resp.Body) + require.Error(t, err, "a truncated chunked body must not read as a clean EOF through the serving chain") + require.NotContains(t, string(body), "PANIC:", "recovery must not write panic text into a committed body") + require.NotContains(t, string(body), "goroutine ", "a stack trace must never reach the client") +} + +// The front streams request bodies, so its chain does not bound them. +func TestAgentChainExemptsAPIBodyLimiter(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("POST /upload", func(w http.ResponseWriter, r *http.Request) { + n, err := io.Copy(io.Discard, r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + fmt.Fprintf(w, "%d", n) + }) + app := newTargetApp(t, mux) + + stack := newEndpointStack(t, agent.EndpointsConfig{}) + stack.startWorker(app.URL, "production", []*livekit.AgentHttp_AgentEndpoint{ + httpEP("/upload", []string{"POST"}, true), + }) + + big := bytes.Repeat([]byte("z"), testMaxAPIBodySize*2) + + resp, err := http.Post(stack.ts.URL+"/agents/test-agent/production/upload", "application/octet-stream", bytes.NewReader(big)) + require.NoError(t, err) + defer resp.Body.Close() + got, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, http.StatusOK, resp.StatusCode) + require.Equal(t, fmt.Sprintf("%d", len(big)), string(got), "the whole body must reach the application") + + // the same body on an API route is still rejected + apiResp, err := http.Post(stack.ts.URL+"/api-sink", "application/octet-stream", bytes.NewReader(big)) + require.NoError(t, err) + defer apiResp.Body.Close() + require.Equal(t, http.StatusRequestEntityTooLarge, apiResp.StatusCode) +} + +// A manifest may declare PUT, so the agent chain's preflight must allow it. +func TestAgentChainCORSAllowsPUT(t *testing.T) { + stack := newEndpointStack(t, agent.EndpointsConfig{}) + + preflight := func(t *testing.T, path string) string { + t.Helper() + req, err := http.NewRequest(http.MethodOptions, stack.ts.URL+path, nil) + require.NoError(t, err) + req.Header.Set("Origin", "https://example.com") + req.Header.Set("Access-Control-Request-Method", http.MethodPut) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + return resp.Header.Get("Access-Control-Allow-Methods") + } + + require.Contains(t, preflight(t, "/agents/test-agent/production/thing"), http.MethodPut) + require.NotContains(t, preflight(t, "/api-sink"), http.MethodPut) +} + +// Disabled leaves the prefix unserved. +func TestAgentEndpointsDisabledIsNotServed(t *testing.T) { + stack := newEndpointStack(t, agent.EndpointsConfig{Disabled: true}) + + resp, err := http.Get(stack.ts.URL + "/agents/test-agent/production/json") + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusNotFound, resp.StatusCode) +} + +// A leading "//" is folded above the split, so it still routes to the agent chain. +func TestAgentChainRoutesDoubleSlashedPath(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("GET /json", func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("ok")) + }) + app := newTargetApp(t, mux) + + stack := newEndpointStack(t, agent.EndpointsConfig{}) + stack.startWorker(app.URL, "production", []*livekit.AgentHttp_AgentEndpoint{ + httpEP("/json", []string{"GET"}, true), + }) + + resp, err := http.Get(stack.ts.URL + "//agents/test-agent/production/json") + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, "ok", string(body)) +} + +// The front resolves access from the grants the api-key auth middleware installs, +// so the agent chain has to carry it. +func TestAgentChainStillResolvesGrants(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("GET /private", func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("ok")) + }) + app := newTargetApp(t, mux) + + stack := newEndpointStack(t, agent.EndpointsConfig{}) + stack.startWorker(app.URL, "production", []*livekit.AgentHttp_AgentEndpoint{ + httpEP("/private", []string{"GET"}, false), + }) + url := stack.ts.URL + "/agents/test-agent/production/private" + + resp, err := http.Get(url) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusUnauthorized, resp.StatusCode, "a non-public route needs a grant") + + req, err := http.NewRequest(http.MethodGet, url, nil) + require.NoError(t, err) + req.Header.Set("Authorization", "Bearer "+stack.endpointToken(t, &auth.AgentEndpointGrant{ + Call: true, AgentName: "test-agent", Deployment: "production", + })) + granted, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer granted.Body.Close() + require.Equal(t, http.StatusOK, granted.StatusCode, "a scoped grant must reach the application") +} diff --git a/pkg/service/agentendpointmiddleware.go b/pkg/service/agentendpointmiddleware.go new file mode 100644 index 000000000..fb153eebe --- /dev/null +++ b/pkg/service/agentendpointmiddleware.go @@ -0,0 +1,57 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "fmt" + "net/http" + "strings" + + "github.com/urfave/negroni/v3" + + "github.com/livekit/protocol/logger" + + "github.com/livekit/livekit-server/pkg/agent/endpoint" +) + +// IsAgentEndpointPath reports whether an escaped request path targets the agent +// endpoint front. The path must already be normalized; see WithPathNormalization. +func IsAgentEndpointPath(escapedPath string) bool { + return strings.HasPrefix(escapedPath, endpoint.PathPrefix) +} + +// AgentRecovery recovers panics, re-panicking http.ErrAbortHandler so it reaches +// net/http and aborts the response. +func AgentRecovery(w http.ResponseWriter, r *http.Request, next http.HandlerFunc) { + defer func() { + rec := recover() + if rec == nil { + return + } + if rec == http.ErrAbortHandler { + panic(rec) + } + err, ok := rec.(error) + if !ok { + err = fmt.Errorf("%v", rec) + } + logger.Errorw("panic serving agent endpoint", err, "path", r.URL.Path) + // a committed response already has its status on the wire + if nrw, ok := w.(negroni.ResponseWriter); !ok || !nrw.Written() { + http.Error(w, "internal server error", http.StatusInternalServerError) + } + }() + next(w, r) +} diff --git a/pkg/service/server.go b/pkg/service/server.go index 8cfadd56c..a69182de9 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -40,7 +40,6 @@ import ( "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils/xtwirp" - "github.com/livekit/livekit-server/pkg/agent/endpoint" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/version" @@ -98,28 +97,6 @@ func NewLivekitServer(conf *config.Config, closedChan: make(chan struct{}), } - middlewares := []negroni.Handler{ - // always first - negroni.NewRecovery(), - // CORS is allowed, we rely on token authentication to prevent improper use - cors.New(cors.Options{ - AllowOriginFunc: func(origin string) bool { - return true - }, - AllowedMethods: []string{"OPTIONS", "HEAD", "GET", "POST", "PATCH", "DELETE"}, - AllowedHeaders: []string{"*"}, - ExposedHeaders: []string{"*"}, - // allow preflight to be cached for a day - MaxAge: 86400, - }), - negroni.HandlerFunc(RemoveDoubleSlashes), - // limit request body size so large messages cannot exhaust memory - NewRequestBodyLimiter(conf.Limit.MaxAPIRequestBodySize), - } - if keyProvider != nil { - middlewares = append(middlewares, NewAPIKeyAuthMiddleware(keyProvider)) - } - serverOptions := []any{ twirp.WithServerHooks(twirp.ChainHooks( TwirpLogger(), @@ -152,11 +129,15 @@ func NewLivekitServer(conf *config.Config, rtcService.SetupRoutes(mux) whipService.SetupRoutes(mux) mux.Handle("/agent", agentService) - mux.Handle(endpoint.PathPrefix, agentService.EndpointFront()) mux.HandleFunc("/", s.defaultHandler) + var agentFront http.Handler + if !conf.Agents.Endpoints.Disabled { + agentFront = agentService.EndpointFront() + } + s.httpServer = &http.Server{ - Handler: configureMiddlewares(mux, middlewares...), + Handler: NewHTTPHandler(conf, keyProvider, mux, agentFront), } if conf.PrometheusPort > 0 { @@ -454,6 +435,58 @@ func (s *LivekitServer) backgroundWorker() { } } +// NewHTTPHandler builds the node's public HTTP handler: the agent endpoint prefix +// on its own middleware chain, everything else on the API chain. A nil agentFront +// leaves the prefix unserved, and those paths fall through to apiHandler. +func NewHTTPHandler(conf *config.Config, keyProvider auth.KeyProvider, apiHandler, agentFront http.Handler) http.Handler { + apiMiddlewares := []negroni.Handler{ + // always first + negroni.NewRecovery(), + // CORS is allowed, we rely on token authentication to prevent improper use + cors.New(corsOptions([]string{"OPTIONS", "HEAD", "GET", "POST", "PATCH", "DELETE"})), + // limit request body size so large messages cannot exhaust memory + NewRequestBodyLimiter(conf.Limit.MaxAPIRequestBodySize), + } + // this chain must not swallow http.ErrAbortHandler or bound the request body + agentMiddlewares := []negroni.Handler{ + negroni.HandlerFunc(AgentRecovery), + // the methods a manifest may declare, less TRACE, which browsers forbid in CORS + cors.New(corsOptions([]string{"OPTIONS", "HEAD", "GET", "POST", "PUT", "PATCH", "DELETE"})), + } + if keyProvider != nil { + authMiddleware := NewAPIKeyAuthMiddleware(keyProvider) + apiMiddlewares = append(apiMiddlewares, authMiddleware) + agentMiddlewares = append(agentMiddlewares, authMiddleware) + } + + api := configureMiddlewares(apiHandler, apiMiddlewares...) + var dispatch http.Handler = api + if agentFront != nil { + agents := configureMiddlewares(agentFront, agentMiddlewares...) + dispatch = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if IsAgentEndpointPath(r.URL.EscapedPath()) { + agents.ServeHTTP(w, r) + return + } + api.ServeHTTP(w, r) + }) + } + return WithPathNormalization(dispatch) +} + +func corsOptions(methods []string) cors.Options { + return cors.Options{ + AllowOriginFunc: func(origin string) bool { + return true + }, + AllowedMethods: methods, + AllowedHeaders: []string{"*"}, + ExposedHeaders: []string{"*"}, + // allow preflight to be cached for a day + MaxAge: 86400, + } +} + func configureMiddlewares(handler http.Handler, middlewares ...negroni.Handler) *negroni.Negroni { n := negroni.New() for _, m := range middlewares { diff --git a/pkg/service/utils.go b/pkg/service/utils.go index cac1b3258..0ee1d55bd 100644 --- a/pkg/service/utils.go +++ b/pkg/service/utils.go @@ -112,6 +112,14 @@ func RemoveDoubleSlashes(w http.ResponseWriter, r *http.Request, next http.Handl next(w, r) } +// WithPathNormalization applies RemoveDoubleSlashes ahead of h, so routing and every +// middleware below it see one path form. +func WithPathNormalization(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + RemoveDoubleSlashes(w, r, h.ServeHTTP) + }) +} + // RequestBodyLimiter bounds the size of an incoming HTTP request body so that // large messages cannot exhaust memory. The Twirp handlers decode the whole // body before any grant check runs, so the limit is applied here, up front.