diff --git a/pkg/agent/endpoint/attempt.go b/pkg/agent/endpoint/attempt.go index 4c1eeaf18..1d1cfa951 100644 --- a/pkg/agent/endpoint/attempt.go +++ b/pkg/agent/endpoint/attempt.go @@ -32,8 +32,9 @@ import ( // attempt is the per-request state shared across worker attempts. type attempt struct { - req *http.Request - path string + req *http.Request + // escPath is the percent-encoded path; a decoded %0A would forge a log line + escPath string // target is the origin-form request target, still percent-encoded target string route *Route diff --git a/pkg/agent/endpoint/front.go b/pkg/agent/endpoint/front.go index 2e042dc87..8acdffa3c 100644 --- a/pkg/agent/endpoint/front.go +++ b/pkg/agent/endpoint/front.go @@ -96,6 +96,18 @@ type Front struct { fallback Fallback // see WithSingleKeyFallback singleKeyFallback bool + // see WithIdentity + identity Identity +} + +// Identity resolves the agent and deployment a request addresses. Reporting +// false leaves them to the URL. +type Identity func(r *http.Request) (agentName, deployment string, ok bool) + +// WithIdentity resolves the agent and deployment from the request itself. +func (f *Front) WithIdentity(fn Identity) *Front { + f.identity = fn + return f } func NewFront(registry *Registry, resolveAccess AccessResolver, log logger.Logger) *Front { @@ -154,6 +166,12 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } agentName, deployment, path, escPath := ep.agentName, ep.deployment, ep.path, ep.escPath + if f.identity != nil { + // must precede resolveAccess, which may consume its source headers + if name, dep, ok := f.identity(r); ok { + agentName, deployment = name, dep + } + } reqID, ok := requestID(r) if !ok { @@ -209,19 +227,9 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { // the registered form. When the request must be relayed, the serving node // runs this same normalization, so no redirect is ever emitted. if route == nil && !partial && !denied { - for _, reg := range candidates { - if alt, ok := reg.Manifest.slashAlternate(path, r.Method); ok { - // a trailing slash is encoding-neutral, so the escaped form - // tracks the alternate directly - if strings.HasSuffix(alt, "/") { - escPath += "/" - } else { - escPath = strings.TrimSuffix(escPath, "/") - } - path = alt - matched, route, partial, denied = matchAll(path) - break - } + if alt, altEsc, ok := slashAlternatePaths(candidates, path, escPath, r.Method); ok { + path, escPath = alt, altEsc + matched, route, partial, denied = matchAll(path) } } if route == nil && f.fallback != nil { @@ -261,7 +269,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { var bodyConsumed atomic.Int64 a := &attempt{ req: r, - path: path, + escPath: escPath, target: requestTarget(escPath, r.URL.RawQuery), route: route, requestID: reqID, @@ -314,6 +322,26 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { f.writeUnavailable(w, "no worker could serve the request") } +// slashAlternatePaths looks for a candidate that serves the trailing-slash +// alternate of path, returning the decoded and escaped forms to retry. +func slashAlternatePaths(candidates []*Registration, path, escPath, method string) (string, string, bool) { + for _, reg := range candidates { + alt, ok := reg.Manifest.slashAlternate(path, method) + if !ok { + continue + } + // %2F decodes to a slash without being a separator, so the alternate + // applies only while the slash is literal in escPath too + switch { + case strings.HasSuffix(alt, "/"): + return alt, escPath + "/", true + case strings.HasSuffix(escPath, "/"): + return alt, strings.TrimSuffix(escPath, "/"), true + } + } + return path, escPath, false +} + // endpointPath is a request split into its routing components. type endpointPath struct { agentName string @@ -352,6 +380,11 @@ func splitEndpointPath(u *url.URL) (endpointPath, error) { if err1 != nil || err2 != nil || err3 != nil { return endpointPath{}, errMalformedPath } + // "_" and "%5F" both land here; neither is a registrable name + // (IsReservedAgentName) + if agentName == UnnamedAgentSegment { + agentName = "" + } return endpointPath{agentName: agentName, deployment: deployment, path: path, escPath: escPath}, nil } @@ -435,7 +468,7 @@ func (f *Front) bridge(w http.ResponseWriter, a *attempt, reg *Registration) (do retryable = a.retryable(err) if !retryable { f.logger.Warnw("agent endpoint request failed", err, - "workerID", reg.WorkerID, "path", a.path, "requestID", a.requestID) + "workerID", reg.WorkerID, "path", a.escPath, "requestID", a.requestID) writeGatewayError(w, err) return true, false } @@ -498,12 +531,12 @@ func (f *Front) logAborted(err error, reg *Registration, a *attempt) { var ce *wire.CompletionError if errors.As(err, &ce) { f.logger.Infow("agent endpoint response aborted", - "workerID", reg.WorkerID, "path", a.path, "requestID", a.requestID, + "workerID", reg.WorkerID, "path", a.escPath, "requestID", a.requestID, "completion", string(ce.Completion), "reason", ce.Reason) return } f.logger.Infow("agent endpoint response aborted", - "workerID", reg.WorkerID, "path", a.path, "requestID", a.requestID, "error", err) + "workerID", reg.WorkerID, "path", a.escPath, "requestID", a.requestID, "error", err) } // completionError normalizes a failure into the protocol's outcome vocabulary. diff --git a/pkg/agent/endpoint/front_test.go b/pkg/agent/endpoint/front_test.go index 062543755..59f1736ce 100644 --- a/pkg/agent/endpoint/front_test.go +++ b/pkg/agent/endpoint/front_test.go @@ -6,6 +6,7 @@ import ( "context" "net/http" "net/http/httptest" + "net/url" "strings" "testing" "time" @@ -240,3 +241,80 @@ func TestFrontDeniedStillRelays(t *testing.T) { require.True(t, got.Credentialed) require.False(t, got.Granted) } + +// the split runs before decoding, so a name or route param may carry any byte +// a percent-encoded segment can hold. +func TestSplitEndpointPath(t *testing.T) { + cases := []struct { + name string + target string + agentName string + deployment string + path string + escPath string + err error + }{ + {name: "plain", target: "/agents/a/d/x", agentName: "a", deployment: "d", path: "/x", escPath: "/x"}, + {name: "no tail", target: "/agents/a/d", agentName: "a", deployment: "d", path: "/", escPath: "/"}, + {name: "spaces", target: "/agents/LODHA%20Vayam%20Agent/d/x", agentName: "LODHA Vayam Agent", deployment: "d", path: "/x", escPath: "/x"}, + {name: "colon and star", target: "/agents/prod%3A%2A/d/x", agentName: "prod:*", deployment: "d", path: "/x", escPath: "/x"}, + {name: "at sign", target: "/agents/charlie%40v1.42.0/d/x", agentName: "charlie@v1.42.0", deployment: "d", path: "/x", escPath: "/x"}, + {name: "brackets", target: "/agents/Nathan%20%5BElara%5D/d/x", agentName: "Nathan [Elara]", deployment: "d", path: "/x", escPath: "/x"}, + {name: "slash in name", target: "/agents/a%2Fb/d/x", agentName: "a/b", deployment: "d", path: "/x", escPath: "/x"}, + {name: "non ascii raw", target: "/agents/agent-ü/d/x", agentName: "agent-ü", deployment: "d", path: "/x", escPath: "/x"}, + {name: "non ascii encoded", target: "/agents/agent-%C3%BC/d/x", agentName: "agent-ü", deployment: "d", path: "/x", escPath: "/x"}, + {name: "past the old 64 byte cap", target: "/agents/" + strings.Repeat("n", 77) + "/d/x", agentName: strings.Repeat("n", 77), deployment: "d", path: "/x", escPath: "/x"}, + {name: "deployment encoded", target: "/agents/a/prod%20us/x", agentName: "a", deployment: "prod us", path: "/x", escPath: "/x"}, + + // escPath keeps the client's encoding; path is what the manifest matches + {name: "encoded tail", target: "/agents/a/d/files/a%2Fb", agentName: "a", deployment: "d", path: "/files/a/b", escPath: "/files/a%2Fb"}, + {name: "escaped percent in tail", target: "/agents/a/d/%2541", agentName: "a", deployment: "d", path: "/%41", escPath: "/%2541"}, + + // "_" addresses the unnamed agent in either form + {name: "bare underscore is unnamed", target: "/agents/_/d/x", agentName: "", deployment: "d", path: "/x", escPath: "/x"}, + {name: "encoded underscore is unnamed", target: "/agents/%5F/d/x", agentName: "", deployment: "d", path: "/x", escPath: "/x"}, + {name: "double encoded underscore is a name", target: "/agents/%255F/d/x", agentName: "%5F", deployment: "d", path: "/x", escPath: "/x"}, + + {name: "not an endpoint path", target: "/other/x", err: errNotEndpointPath}, + {name: "no deployment segment", target: "/agents/a", err: errNotEndpointPath}, + {name: "empty name", target: "/agents//d/x", err: errNotEndpointPath}, + {name: "empty deployment", target: "/agents/a//x", err: errNotEndpointPath}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + u, err := url.ParseRequestURI(c.target) + require.NoError(t, err) + + ep, err := splitEndpointPath(u) + if c.err != nil { + require.ErrorIs(t, err, c.err) + return + } + require.NoError(t, err) + require.Equal(t, c.agentName, ep.agentName) + require.Equal(t, c.deployment, ep.deployment) + require.Equal(t, c.path, ep.path) + require.Equal(t, c.escPath, ep.escPath) + }) + } +} + +// a trailing %2F decodes to a slash without being a separator, so the +// trailing-slash alternate does not apply to it. +func TestFrontSlashAlternateIgnoresEncodedSlash(t *testing.T) { + f := fallbackFront(t, nil, true) + + // 503 is dispatch reached: the fake session opens no stream + require.Equal(t, http.StatusServiceUnavailable, serveFront(f, "/known/").Code) + require.Equal(t, http.StatusNotFound, serveFront(f, "/known%2F").Code) +} + +func TestIsReservedAgentName(t *testing.T) { + for _, n := range []string{"_", ".", ".."} { + require.True(t, IsReservedAgentName(n), n) + } + for _, n := range []string{"", "a", "_x", "x_", "...", "LODHA Vayam Agent", "%5F"} { + require.False(t, IsReservedAgentName(n), n) + } +} diff --git a/pkg/agent/endpoint/registry.go b/pkg/agent/endpoint/registry.go index 41c562364..95f74d95e 100644 --- a/pkg/agent/endpoint/registry.go +++ b/pkg/agent/endpoint/registry.go @@ -25,6 +25,17 @@ import ( // empty deployment name (self-hosted workers typically set none). const DefaultDeployment = "default" +// UnnamedAgentSegment is the URL segment that addresses workers registered with +// an empty agent name. +const UnnamedAgentSegment = "_" + +// IsReservedAgentName reports whether a name cannot address a worker. These are +// unreserved characters (RFC 3986 §2.3), so percent-encoding them yields no +// distinct form to address them by. +func IsReservedAgentName(agentName string) bool { + return agentName == UnnamedAgentSegment || agentName == "." || agentName == ".." +} + // DefaultMaxStreams is the soft per-session concurrency cap used only for // capacity weighting; QUIC's own stream limit is the hard bound. const DefaultMaxStreams = 256 diff --git a/pkg/agent/worker.go b/pkg/agent/worker.go index c9a8e5983..b2e55db05 100644 --- a/pkg/agent/worker.go +++ b/pkg/agent/worker.go @@ -25,6 +25,7 @@ import ( "go.uber.org/multierr" "google.golang.org/protobuf/proto" + "github.com/livekit/livekit-server/pkg/agent/endpoint" protoagent "github.com/livekit/protocol/agent" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" @@ -254,13 +255,13 @@ func (h *WorkerRegisterer) HandleRegister(req *livekit.RegisterWorkerRequest) er if h.endpointSettings == nil { return errors.New("agent HTTP endpoints are not supported by this server") } - // endpoints are addressed at /agents/{agent_name}/{deployment}/...; a - // worker serving them needs a non-empty, URL-safe agent name + // endpoints are addressed at /agents/{agent_name}/{deployment}/..., with + // the name percent-encoded into that segment if req.GetAgentName() == "" { return errors.New("agent HTTP endpoints require an agent name") } - if err := protoagent.ValidateAgentName(req.GetAgentName()); err != nil { - return err + if endpoint.IsReservedAgentName(req.GetAgentName()) { + return fmt.Errorf("agent name %q is reserved and cannot serve HTTP endpoints", req.GetAgentName()) } settings, err := h.endpointSettings(req) if err != nil { diff --git a/pkg/agent/worker_test.go b/pkg/agent/worker_test.go new file mode 100644 index 000000000..b89000669 --- /dev/null +++ b/pkg/agent/worker_test.go @@ -0,0 +1,77 @@ +// Copyright 2026 LiveKit, Inc. + +package agent_test + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/agent" + "github.com/livekit/protocol/livekit" +) + +type nopSignalConn struct{} + +func (nopSignalConn) WriteServerMessage(*livekit.ServerMessage) (int, error) { return 0, nil } +func (nopSignalConn) ReadWorkerMessage() (*livekit.WorkerMessage, int, error) { return nil, 0, nil } +func (nopSignalConn) SetReadDeadline(time.Time) error { return nil } +func (nopSignalConn) Close() error { return nil } +func (nopSignalConn) CloseWithReason(string) error { return nil } + +// the name is percent-encoded into a URL path segment. +func TestHandleRegisterEndpointAgentNames(t *testing.T) { + cases := []struct { + name string + agentName string + ok bool + }{ + {"plain", "test-agent", true}, + {"underscore", "my_agent", true}, + {"spaces", "LODHA Vayam Agent", true}, + {"colon and star", "prod:*", true}, + {"at sign", "charlie@v1.42.0", true}, + {"slash", "models/gemma/v2", true}, + {"brackets", "Nathan [Elara Agent]", true}, + {"non ascii", "Sehhaty الصحة", true}, + {"past the old 64 byte cap", strings.Repeat("n", 77), true}, + + {"empty", "", false}, + {"unnamed sentinel", "_", false}, + {"dot", ".", false}, + {"dot dot", "..", false}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + h := agent.NewWorkerRegisterer(nopSignalConn{}, &livekit.ServerInfo{}, agent.WorkerRegistration{}). + WithEndpointSettings(func(*livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) { + return &livekit.AgentHttp_AgentEndpointSettings{}, nil + }) + + err := h.HandleRegister(&livekit.RegisterWorkerRequest{ + Type: livekit.JobType_JT_ROOM, + AgentName: c.agentName, + Endpoints: []*livekit.AgentHttp_AgentEndpoint{{Path: "/hook", Methods: []string{"GET"}}}, + }) + if c.ok { + require.NoError(t, err) + } else { + require.Error(t, err) + } + }) + } +} + +// the reserved names constrain only workers that declare endpoints. +func TestHandleRegisterWithoutEndpointsIgnoresReservedNames(t *testing.T) { + for _, name := range []string{"", "_", ".", ".."} { + h := agent.NewWorkerRegisterer(nopSignalConn{}, &livekit.ServerInfo{}, agent.WorkerRegistration{}) + require.NoError(t, h.HandleRegister(&livekit.RegisterWorkerRequest{ + Type: livekit.JobType_JT_ROOM, + AgentName: name, + }), name) + } +} diff --git a/pkg/service/agentendpoint_test.go b/pkg/service/agentendpoint_test.go index 45b87d0ee..b9c59343f 100644 --- a/pkg/service/agentendpoint_test.go +++ b/pkg/service/agentendpoint_test.go @@ -31,6 +31,7 @@ import ( "net" "net/http" "net/http/httptest" + "net/url" "strings" "sync/atomic" "testing" @@ -121,11 +122,15 @@ func newEndpointStack(t *testing.T, endpointsCfg agent.EndpointsConfig) *endpoin } func (s *endpointStack) startWorker(target string, deployment string, endpoints []*livekit.AgentHttp_AgentEndpoint) *conformance.Worker { + return s.startNamedWorker("test-agent", target, deployment, endpoints) +} + +func (s *endpointStack) startNamedWorker(agentName, target, deployment string, endpoints []*livekit.AgentHttp_AgentEndpoint) *conformance.Worker { w := conformance.New(conformance.Config{ ServerURL: s.wtURL, APIKey: testKey, APISecret: testSecret, - AgentName: "test-agent", + AgentName: agentName, Deployment: deployment, Endpoints: endpoints, TargetAddr: strings.TrimPrefix(target, "http://"), @@ -592,3 +597,56 @@ func TestAgentEndpointsNonUTF8HeaderSurvives(t *testing.T) { require.Equal(t, "data", string(body)) require.Equal(t, disposition, resp.Header.Get("Content-Disposition")) } + +// the agent name and the request tail are percent-encoded path segments, and +// the worker's request line carries the client's bytes. +func TestAgentEndpointsEncodedNameAndPath(t *testing.T) { + const agentName = "LODHA Vayam/Agent" + + targets := make(chan string, 4) + mux := http.NewServeMux() + mux.HandleFunc("GET /echo/{rest...}", func(w http.ResponseWriter, r *http.Request) { + targets <- r.RequestURI + _, _ = w.Write([]byte("ok")) + }) + app := newTargetApp(t, mux) + + stack := newEndpointStack(t, agent.EndpointsConfig{}) + stack.startNamedWorker(agentName, app.URL, "production", []*livekit.AgentHttp_AgentEndpoint{ + httpEP("/echo/{rest:path}", []string{"GET"}, true), + }) + + base := stack.ts.URL + "/agents/" + url.PathEscape(agentName) + "/production" + + get := func(t *testing.T, path string) (int, string) { + t.Helper() + resp, err := http.Get(base + path) + require.NoError(t, err) + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + select { + case target := <-targets: + return resp.StatusCode, target + case <-time.After(5 * time.Second): + return resp.StatusCode, "" + } + } + + t.Run("a name with a space and a slash addresses its worker", func(t *testing.T) { + code, target := get(t, "/echo/plain") + require.Equal(t, 200, code) + require.Equal(t, "/echo/plain", target) + }) + + t.Run("an encoded slash in the tail stays encoded", func(t *testing.T) { + code, target := get(t, "/echo/a%2Fb") + require.Equal(t, 200, code) + require.Equal(t, "/echo/a%2Fb", target, "a decoded %2F would re-emit as a separator and change the resource") + }) + + t.Run("the query is passed through verbatim", func(t *testing.T) { + code, target := get(t, "/echo/q?a=1&b=%2F%20x") + require.Equal(t, 200, code) + require.Equal(t, "/echo/q?a=1&b=%2F%20x", target) + }) +} diff --git a/pkg/service/utils.go b/pkg/service/utils.go index cc31be859..cac1b3258 100644 --- a/pkg/service/utils.go +++ b/pkg/service/utils.go @@ -101,8 +101,13 @@ func boolValue(s string) bool { } func RemoveDoubleSlashes(w http.ResponseWriter, r *http.Request, next http.HandlerFunc) { - if strings.HasPrefix(r.URL.Path, "//") { + // Path and RawPath must move together: once they disagree, EscapedPath() + // re-encodes from Path and every escape in the request is lost. + if strings.HasPrefix(r.URL.EscapedPath(), "//") { r.URL.Path = r.URL.Path[1:] + if r.URL.RawPath != "" { + r.URL.RawPath = r.URL.RawPath[1:] + } } next(w, r) } diff --git a/pkg/service/utils_test.go b/pkg/service/utils_test.go index 1f675c6f8..2c1a43060 100644 --- a/pkg/service/utils_test.go +++ b/pkg/service/utils_test.go @@ -19,6 +19,7 @@ import ( "compress/gzip" "context" "net/http" + "net/http/httptest" "testing" "time" @@ -127,3 +128,30 @@ func TestDecompressGzip(t *testing.T) { require.Contains(t, err.Error(), "cannot read decompressed") }) } + +// URL.Path and URL.RawPath must stay in step: once they disagree, +// EscapedPath() re-encodes from Path and every escape is lost. +func TestRemoveDoubleSlashes(t *testing.T) { + cases := []struct { + name string + target string + want string + }{ + {"doubled slash is trimmed", "//agents/a/d/x", "/agents/a/d/x"}, + {"single slash untouched", "/agents/a/d/x", "/agents/a/d/x"}, + {"doubled slash keeps the encoded tail", "//agents/a/d/f%2Fg", "/agents/a/d/f%2Fg"}, + {"encoded name survives", "//agents/LODHA%20Vayam/d/x", "/agents/LODHA%20Vayam/d/x"}, + {"leading encoded slash is not doubled", "/%2Fagents/a/d/x", "/%2Fagents/a/d/x"}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + r := httptest.NewRequest(http.MethodGet, c.target, nil) + var got string + service.RemoveDoubleSlashes(httptest.NewRecorder(), r, func(_ http.ResponseWriter, r *http.Request) { + got = r.URL.EscapedPath() + }) + require.Equal(t, c.want, got) + }) + } +}