From 9f73fde0d419191fdf785d741e9d095fefacdff8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Wed, 7 Oct 2026 20:14:03 -0700 Subject: [PATCH] agent endpoint: the thread grant, scoped to the agent's endpoints livekit/protocol at the head of PR 1843: the agent session is a thread. The preamble carries thread_id and thread_grant, the forged header is X-LiveKit-Thread-Grant, and a grant that lists endpoints opens only those: the first path segment under the deployment, 403 outside it before any worker sees the request. The local protocol replace goes with the pin. --- go.mod | 6 +- go.sum | 8 +-- pkg/agent/endpoint/attempt.go | 40 +++++------ pkg/agent/endpoint/front.go | 77 ++++++++++++--------- pkg/agent/endpoint/front_test.go | 111 +++++++++++++++++++------------ pkg/service/agentendpoint.go | 26 +++++++- 6 files changed, 163 insertions(+), 105 deletions(-) diff --git a/go.mod b/go.mod index 915aad1e1..1666481f1 100644 --- a/go.mod +++ b/go.mod @@ -21,8 +21,8 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095 - github.com/livekit/protocol v1.52.2-0.20261006063934-0761f135e913 - github.com/livekit/psrpc v0.8.1 + github.com/livekit/protocol v1.52.2-0.20261008023442-60589e83886b + github.com/livekit/psrpc v0.8.2 github.com/mackerelio/go-osstat v0.2.8 github.com/magefile/mage v1.17.2 github.com/mitchellh/go-homedir v1.1.0 @@ -156,7 +156,7 @@ require ( golang.org/x/tools v0.51.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20261005182115-fad411399dd8 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8 // indirect - google.golang.org/grpc v1.84.0 // indirect + google.golang.org/grpc v1.84.0 ) tool ( diff --git a/go.sum b/go.sum index 16ddea2ec..8ea6e4cf3 100644 --- a/go.sum +++ b/go.sum @@ -166,10 +166,10 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095 h1:BcliKAXoMhl/nWmzQweQ5kmh4Qqagxl4s3Z5pvM/7AY= github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.52.2-0.20261006063934-0761f135e913 h1:LopYipWFeZbKrjRFSYgX7GBJx6FBQ2mvtcoZU4+ynJM= -github.com/livekit/protocol v1.52.2-0.20261006063934-0761f135e913/go.mod h1:EJRuXu4hCsje+SiDnPGJ9U2wpFbwTl6bzC4ZxPSdRec= -github.com/livekit/psrpc v0.8.1 h1:aeEnC2WRWB2tzNpmrifog8B+DD9+0K2Fk8xT/0kMuxg= -github.com/livekit/psrpc v0.8.1/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q= +github.com/livekit/protocol v1.52.2-0.20261008023442-60589e83886b h1:BX9CZ2H6evJPiyr9l7mz3zQhZk0u5vIuTWj6v7kPWh4= +github.com/livekit/protocol v1.52.2-0.20261008023442-60589e83886b/go.mod h1:k7v95/DNhMkslMeBxdQ+lQ6gKc8hMNGYB7t0+6sEGU4= +github.com/livekit/psrpc v0.8.2 h1:7tbkhBJoYlK5Csfkdv1XYbZjyAQeaEZgupvIXNCv7Dc= +github.com/livekit/psrpc v0.8.2/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q= github.com/livekit/webrtc-pion/v4 v4.2.18-warp.6 h1:UaGYlQ/Ma73LN12xPDlH3OBYNfOEDTVOtHdRZbHg2dY= github.com/livekit/webrtc-pion/v4 v4.2.18-warp.6/go.mod h1:Kxgo6xwFzVK38XVlQTBT1zj2JU5G7R0R0pa5w5d9d/Q= github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4= diff --git a/pkg/agent/endpoint/attempt.go b/pkg/agent/endpoint/attempt.go index 57e3e34db..8ed9d21ef 100644 --- a/pkg/agent/endpoint/attempt.go +++ b/pkg/agent/endpoint/attempt.go @@ -34,10 +34,10 @@ import ( "github.com/livekit/protocol/utils/guid" ) -// sessionGrantHeader is the header a worker reached without a node accepts the +// threadGrantHeader is the header a worker reached without a node accepts the // grant on. Through a node the grant travels in the preamble, so this header is // reserved to the node and stripped from every forwarded request. -const sessionGrantHeader = "X-LiveKit-Session-Grant" +const threadGrantHeader = "X-LiveKit-Thread-Grant" // attempt is the per-request state shared across worker attempts. type attempt struct { @@ -48,15 +48,15 @@ type attempt struct { target string requestID string granted bool - // grant is the caller's verified agent session grant, nil when the token + // grant is the caller's verified agent thread grant, nil when the token // carried none. Every kind of route is told: the worker enforces it. - grant *livekit.AgentHttp_AgentSessionGrant - // sessionID is the agent session the front placed the request in; empty - // when it placed none. Only a text-kind declaration is told. - sessionID string - body io.Reader - pools *bridgePools - preamble *livekit.AgentHttp_StreamPreamble + grant *livekit.AgentHttp_AgentThreadGrant + // threadID is the thread the front placed the request in; empty when it + // placed none. Only a text-kind declaration is told. + threadID string + body io.Reader + pools *bridgePools + preamble *livekit.AgentHttp_StreamPreamble // the serialized canonical request head, built once and replayed verbatim head []byte // client body bytes consumed when this attempt began @@ -107,11 +107,11 @@ func (a *attempt) newPreamble() *livekit.AgentHttp_StreamPreamble { clientAddr, _, _ := net.SplitHostPort(a.req.RemoteAddr) p := &livekit.AgentHttp_StreamPreamble{ - RequestId: a.requestID, - Authorized: a.granted, - SessionGrant: a.grant, - ClientAddr: clientAddr, - Scheme: scheme, + RequestId: a.requestID, + Authorized: a.granted, + ThreadGrant: a.grant, + ClientAddr: clientAddr, + Scheme: scheme, } return p } @@ -123,9 +123,9 @@ func (a *attempt) newPreamble() *livekit.AgentHttp_StreamPreamble { func (a *attempt) setRoute(w routeWorker) { a.preamble.Route = w.raw a.preamble.Kind = w.kind - a.preamble.SessionId = "" + a.preamble.ThreadId = "" if w.kind == livekit.AgentHttp_AEK_TEXT { - a.preamble.SessionId = a.sessionID + a.preamble.ThreadId = a.threadID } } @@ -141,15 +141,15 @@ func (a *attempt) requestHeader() http.Header { h.Set("Connection", "Upgrade") h.Set("Upgrade", "websocket") } - // identity travels as preamble.authorized and preamble.session_grant; the + // identity travels as preamble.authorized and preamble.thread_grant; the // token must never reach the worker if isLiveKitBearer(h.Get("Authorization")) { h.Del("Authorization") } - // the session grant is the node's to assert, from the verified token. A + // the thread grant is the node's to assert, from the verified token. A // worker reached directly, with no node in front, may still read this // header, so a client must not be able to plant one here. - h.Del(sessionGrantHeader) + h.Del(threadGrantHeader) // the signalling namespace is the node's to write: a client able to set // x-lk-completion could tell the worker a body it cut short was whole wire.StripReservedHeaders(h) diff --git a/pkg/agent/endpoint/front.go b/pkg/agent/endpoint/front.go index 770e1fa61..6c02bf2d6 100644 --- a/pkg/agent/endpoint/front.go +++ b/pkg/agent/endpoint/front.go @@ -61,8 +61,8 @@ const ( // worker cannot hold a client open by trickling them forever. maxInformationalHeads = 8 - // DefaultSessionPeek bounds the request body prefix a SessionResolver sees. - DefaultSessionPeek = 4 << 10 + // DefaultThreadPeek bounds the request body prefix a ThreadResolver sees. + DefaultThreadPeek = 4 << 10 maxResponseHeadSize = 1 << 20 responseBufSize = 8 << 10 @@ -119,10 +119,11 @@ type Access struct { // curried on the deployment it resolved. nil means local misses are final. Fallback Fallback Level AccessLevel - // SessionGrant is the verified token's agentSession claim, nil when the + // ThreadGrant is the verified token's agentThread claim, nil when the // token carried none or there was no token. It reaches the worker in the - // preamble; the token itself never does. - SessionGrant *auth.AgentSessionGrant + // preamble; the token itself never does. Its endpoints, when it lists any, + // are checked here: a route outside them is forbidden. + ThreadGrant *auth.AgentThreadGrant } // AccessResolver maps an inbound request, plus the agent and deployment its URL @@ -141,15 +142,15 @@ type Front struct { // false leaves them to the URL. type Identity func(r *http.Request) (agentName, deployment string, ok bool) -// SessionResolver places a request on an AEK_TEXT route in an agent session. -// It runs once per request, after access is resolved and before a worker is -// picked, and only when a matched declaration is of the text kind: that kind is -// the worker's consent to the front reading its body. body is the first bytes -// of the request body, at most SessionPeek; the worker receives the whole body -// regardless. Reporting false leaves the session to the worker, which resolves -// it from the body itself. Cloud installs one backed by the session registry; +// ThreadResolver places a request on an AEK_TEXT route in a thread. It runs +// once per request, after access is resolved and before a worker is picked, +// and only when a matched declaration is of the text kind: that kind is the +// worker's consent to the front reading its body. body is the first bytes of +// the request body, at most ThreadPeek; the worker receives the whole body +// regardless. Reporting false leaves the thread to the worker, which resolves +// it from the body itself. Cloud installs one backed by the threads service; // OSS leaves it nil, and a text route is then forwarded unchanged. -type SessionResolver func(r *http.Request, agentName, deployment string, body []byte) (sessionID string, ok bool) +type ThreadResolver func(r *http.Request, agentName, deployment string, body []byte) (threadID string, ok bool) // FrontParams configures a Front. Fields are read on every request once the // Front is serving, so none may change after construction. @@ -158,12 +159,12 @@ type FrontParams struct { Logger logger.Logger Identity Identity - // ResolveSession places text-kind requests in an agent session. nil - // forwards them unchanged. - ResolveSession SessionResolver - // SessionPeek bounds the body prefix ResolveSession sees. 0 means - // DefaultSessionPeek. - SessionPeek int + // ResolveThread places text-kind requests in a thread. nil forwards them + // unchanged. + ResolveThread ThreadResolver + // ThreadPeek bounds the body prefix ResolveThread sees. 0 means + // DefaultThreadPeek. + ThreadPeek int // MaxTunnelsPerWorker caps one worker's concurrent WebSocket tunnels. 0 // means half its session's MaxStreams. @@ -278,9 +279,16 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - var sessionID string - if f.params.ResolveSession != nil && declaresText(matched) { - peek, err := peekBody(r, f.sessionPeek()) + // a grant that lists endpoints covers this one or none: the endpoint is the + // first segment of the path under the deployment + if access.ThreadGrant != nil && !access.ThreadGrant.AllowsEndpoint(endpointOf(path)) { + http.Error(w, "forbidden", http.StatusForbidden) + return + } + + var threadID string + if f.params.ResolveThread != nil && declaresText(matched) { + peek, err := peekBody(r, f.threadPeek()) if err != nil { // the client's body failed before any worker saw the request, so // there is nothing to place and nothing to retry @@ -288,8 +296,8 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.Error(w, "bad request", http.StatusBadRequest) return } - if id, ok := f.params.ResolveSession(r, agentName, deployment, peek); ok { - sessionID = id + if id, ok := f.params.ResolveThread(r, agentName, deployment, peek); ok { + threadID = id } } @@ -300,8 +308,8 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { target: requestTarget(escPath, r.URL.RawQuery), requestID: reqID, granted: granted, - grant: access.SessionGrant.ToProto(), - sessionID: sessionID, + grant: access.ThreadGrant.ToProto(), + threadID: threadID, webSocket: isWebSocketUpgrade(r), pools: f.pools, } @@ -363,15 +371,22 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { f.writeUnavailable(w, "no worker could serve the request") } -func (f *Front) sessionPeek() int { - if f.params.SessionPeek > 0 { - return f.params.SessionPeek +func (f *Front) threadPeek() int { + if f.params.ThreadPeek > 0 { + return f.params.ThreadPeek } - return DefaultSessionPeek + return DefaultThreadPeek +} + +// endpointOf is the endpoint a decoded path addresses, its first segment: the +// name every door of the agent is reached under, {agent_name}/{endpoint}. +func endpointOf(path string) string { + endpoint, _, _ := strings.Cut(strings.TrimPrefix(path, "/"), "/") + return endpoint } // declaresText reports whether any declaration the request matched is of the -// text kind. Declarers of one route may disagree, so the session is resolved +// text kind. Declarers of one route may disagree, so the thread is resolved // once and handed only to the workers that asked for it (setRoute). func declaresText(matched []routeWorker) bool { for _, w := range matched { diff --git a/pkg/agent/endpoint/front_test.go b/pkg/agent/endpoint/front_test.go index adf362100..9ca608faf 100644 --- a/pkg/agent/endpoint/front_test.go +++ b/pkg/agent/endpoint/front_test.go @@ -473,16 +473,16 @@ func postText(t *testing.T, ts *httptest.Server, path, body string) { } // the preamble carries the kind the picked worker declared the route with, and -// with no resolver installed a text route is forwarded unchanged: no session, +// with no resolver installed a text route is forwarded unchanged: no thread, // whole body. func TestFrontTextRouteWithoutResolverPassesThrough(t *testing.T) { ts, seen := textFront(t, FrontParams{}) - body := strings.Repeat("x", 3*DefaultSessionPeek) + body := strings.Repeat("x", 3*DefaultThreadPeek) postText(t, ts, "/a2a", body) got := <-seen require.Equal(t, livekit.AgentHttp_AEK_TEXT, got.pre.GetKind()) - require.Empty(t, got.pre.GetSessionId()) + require.Empty(t, got.pre.GetThreadId()) require.Equal(t, body, got.body) postText(t, ts, "/plain", body) @@ -494,25 +494,25 @@ func TestFrontTextRouteWithoutResolverPassesThrough(t *testing.T) { // a resolver sees the request head and a bounded prefix of the body; its id // reaches the text-kind worker in the preamble, and the worker still gets the // whole body. -func TestFrontSessionResolverPlacesTextRoute(t *testing.T) { +func TestFrontThreadResolverPlacesTextRoute(t *testing.T) { type call struct { method, path, channel, agent, deployment string peek string } calls := make(chan call, 1) ts, seen := textFront(t, FrontParams{ - SessionPeek: 8, - ResolveSession: func(r *http.Request, agentName, deployment string, body []byte) (string, bool) { + ThreadPeek: 8, + ResolveThread: func(r *http.Request, agentName, deployment string, body []byte) (string, bool) { calls <- call{r.Method, r.URL.Path, r.Header.Get("X-Channel"), agentName, deployment, string(body)} return "AT_1", true }, }) - body := `{"jsonrpc":"2.0","params":` + strings.Repeat("x", 2*DefaultSessionPeek) + `}` + body := `{"jsonrpc":"2.0","params":` + strings.Repeat("x", 2*DefaultThreadPeek) + `}` postText(t, ts, "/a2a", body) got := <-seen require.Equal(t, livekit.AgentHttp_AEK_TEXT, got.pre.GetKind()) - require.Equal(t, "AT_1", got.pre.GetSessionId()) + require.Equal(t, "AT_1", got.pre.GetThreadId()) require.Equal(t, body, got.body, "the peek must not take bytes from the worker") c := <-calls @@ -522,29 +522,29 @@ func TestFrontSessionResolverPlacesTextRoute(t *testing.T) { postText(t, ts, "/plain", body) got = <-seen require.Equal(t, livekit.AgentHttp_AEK_HTTP, got.pre.GetKind()) - require.Empty(t, got.pre.GetSessionId()) + require.Empty(t, got.pre.GetThreadId()) require.Equal(t, body, got.body) require.Empty(t, calls) } -// a resolver that places nothing leaves the session to the worker. -func TestFrontSessionResolverDeclines(t *testing.T) { +// a resolver that places nothing leaves the thread to the worker. +func TestFrontThreadResolverDeclines(t *testing.T) { ts, seen := textFront(t, FrontParams{ - ResolveSession: func(*http.Request, string, string, []byte) (string, bool) { return "", false }, + ResolveThread: func(*http.Request, string, string, []byte) (string, bool) { return "", false }, }) postText(t, ts, "/a2a", "hello") got := <-seen require.Equal(t, livekit.AgentHttp_AEK_TEXT, got.pre.GetKind()) - require.Empty(t, got.pre.GetSessionId()) + require.Empty(t, got.pre.GetThreadId()) require.Equal(t, "hello", got.body) } // a body shorter than the peek bound is handed over whole, with nothing lost to // the read-ahead; an absent body peeks nothing. -func TestFrontSessionPeekShortBody(t *testing.T) { +func TestFrontThreadPeekShortBody(t *testing.T) { peeks := make(chan string, 1) ts, seen := textFront(t, FrontParams{ - ResolveSession: func(_ *http.Request, _, _ string, body []byte) (string, bool) { + ResolveThread: func(_ *http.Request, _, _ string, body []byte) (string, bool) { peeks <- string(body) return "AT_2", true }, @@ -553,13 +553,13 @@ func TestFrontSessionPeekShortBody(t *testing.T) { postText(t, ts, "/a2a", "short") require.Equal(t, "short", <-peeks) got := <-seen - require.Equal(t, "AT_2", got.pre.GetSessionId()) + require.Equal(t, "AT_2", got.pre.GetThreadId()) require.Equal(t, "short", got.body) postText(t, ts, "/a2a", "") require.Empty(t, <-peeks) got = <-seen - require.Equal(t, "AT_2", got.pre.GetSessionId()) + require.Equal(t, "AT_2", got.pre.GetThreadId()) require.Empty(t, got.body) } @@ -568,7 +568,7 @@ const testAPIKey, testAPISecret = "APIfront", "front-secret-front-secret-front-s // verifiedAccess resolves access the way the server's middleware and resolver // do together: a LiveKit token from the bearer, or from access_token on the // query when there is no bearer, is verified and its claims decide the level -// and the session grant. No token is anonymous. +// and the thread grant. No token is anonymous. func verifiedAccess(agentName, deployment string) AccessResolver { return func(r *http.Request, _, _ string) (Access, bool) { raw := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") @@ -590,7 +590,7 @@ func verifiedAccess(agentName, deployment string) AccessResolver { if claims.AgentEndpoint.Allows(agentName, deployment) { level = AccessGranted } - return Access{Level: level, SessionGrant: claims.AgentSession}, true + return Access{Level: level, ThreadGrant: claims.AgentThread}, true } } @@ -603,23 +603,23 @@ func withScope(resolve AccessResolver, scope *Scope) AccessResolver { } } -func mintToken(t *testing.T, grant *auth.AgentSessionGrant) string { +func mintToken(t *testing.T, grant *auth.AgentThreadGrant) string { t.Helper() tok, err := auth.NewAccessToken(testAPIKey, testAPISecret). SetIdentity("caller"). SetAgentEndpointGrant(&auth.AgentEndpointGrant{Call: true}). - SetAgentSessionGrant(grant). + SetAgentThreadGrant(grant). ToJWT() require.NoError(t, err) return tok } -// the verified token's agentSession claim reaches the worker in the preamble, on +// the verified token's agentThread claim reaches the worker in the preamble, on // a text route and a plain one alike, while the token itself does not. -func TestFrontSessionGrantTravelsInPreamble(t *testing.T) { +func TestFrontThreadGrantTravelsInPreamble(t *testing.T) { ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")}) - grant := &auth.AgentSessionGrant{ - AgentName: "a", Sub: "demo-user", SessionIDs: []string{"AT_1", "AT_2"}, + grant := &auth.AgentThreadGrant{ + AgentName: "a", Sub: "demo-user", ThreadIDs: []string{"AT_1", "AT_2"}, List: true, Create: true, Send: true, Read: true, } tok := mintToken(t, grant) @@ -636,23 +636,23 @@ func TestFrontSessionGrantTravelsInPreamble(t *testing.T) { got := <-seen require.True(t, got.pre.GetAuthorized()) require.Empty(t, got.header.Get("Authorization"), "the token must never reach the worker") - require.Equal(t, "a", got.pre.GetSessionGrant().GetAgentName()) - require.Equal(t, "demo-user", got.pre.GetSessionGrant().GetSub()) - require.Equal(t, []string{"AT_1", "AT_2"}, got.pre.GetSessionGrant().GetSessionIds()) - require.True(t, got.pre.GetSessionGrant().GetList()) - require.True(t, got.pre.GetSessionGrant().GetCreate()) - require.True(t, got.pre.GetSessionGrant().GetSend()) - require.True(t, got.pre.GetSessionGrant().GetRead()) - require.False(t, got.pre.GetSessionGrant().GetDelete()) + require.Equal(t, "a", got.pre.GetThreadGrant().GetAgentName()) + require.Equal(t, "demo-user", got.pre.GetThreadGrant().GetSub()) + require.Equal(t, []string{"AT_1", "AT_2"}, got.pre.GetThreadGrant().GetThreadIds()) + require.True(t, got.pre.GetThreadGrant().GetList()) + require.True(t, got.pre.GetThreadGrant().GetCreate()) + require.True(t, got.pre.GetThreadGrant().GetSend()) + require.True(t, got.pre.GetThreadGrant().GetRead()) + require.False(t, got.pre.GetThreadGrant().GetDelete()) } } // a browser WebSocket cannot set headers, so the token may ride the query as // access_token: it is verified the same way, the grant travels in the preamble, // and the query the worker sees no longer carries it. -func TestFrontSessionGrantFromQueryToken(t *testing.T) { +func TestFrontThreadGrantFromQueryToken(t *testing.T) { ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")}) - tok := mintToken(t, &auth.AgentSessionGrant{AgentName: "a", Sub: "demo-user", Send: true}) + tok := mintToken(t, &auth.AgentThreadGrant{AgentName: "a", Sub: "demo-user", Send: true}) req, err := http.NewRequest(http.MethodPost, ts.URL+PathPrefix+"a/d/a2a?access_token="+tok+"&keep=1", strings.NewReader("hello")) require.NoError(t, err) @@ -662,23 +662,23 @@ func TestFrontSessionGrantFromQueryToken(t *testing.T) { require.Equal(t, http.StatusNoContent, resp.StatusCode) got := <-seen - require.Equal(t, "demo-user", got.pre.GetSessionGrant().GetSub()) - require.Equal(t, "a", got.pre.GetSessionGrant().GetAgentName()) - require.True(t, got.pre.GetSessionGrant().GetSend()) + require.Equal(t, "demo-user", got.pre.GetThreadGrant().GetSub()) + require.Equal(t, "a", got.pre.GetThreadGrant().GetAgentName()) + require.True(t, got.pre.GetThreadGrant().GetSend()) require.Empty(t, got.header.Get("Authorization")) require.Equal(t, "/a2a?keep=1", got.target, "the query token is stripped like the bearer") } -// the grant is the node's to assert: a client-set X-LiveKit-Session-Grant is -// dropped, and a token with no agentSession claim leaves the field unset. -func TestFrontDropsForgedSessionGrantHeader(t *testing.T) { +// the grant is the node's to assert: a client-set X-LiveKit-Thread-Grant is +// dropped, and a token with no agentThread claim leaves the field unset. +func TestFrontDropsForgedThreadGrantHeader(t *testing.T) { ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")}) tok := mintToken(t, nil) req, err := http.NewRequest(http.MethodPost, ts.URL+PathPrefix+"a/d/a2a", strings.NewReader("hello")) require.NoError(t, err) req.Header.Set("Authorization", "Bearer "+tok) - req.Header.Set(sessionGrantHeader, `{"agentName":"a","sub":"someone-else","send":true}`) + req.Header.Set(threadGrantHeader, `{"agentName":"a","sub":"someone-else","send":true}`) req.Header.Set("X-Keep", "1") resp, err := http.DefaultClient.Do(req) require.NoError(t, err) @@ -687,8 +687,31 @@ func TestFrontDropsForgedSessionGrantHeader(t *testing.T) { got := <-seen require.True(t, got.pre.GetAuthorized()) - require.Nil(t, got.pre.GetSessionGrant(), "a project token with no agentSession claim grants no session") - require.Empty(t, got.header.Values(sessionGrantHeader), "a client cannot forge the grant") + require.Nil(t, got.pre.GetThreadGrant(), "a project token with no agentThread claim grants no thread") + require.Empty(t, got.header.Values(threadGrantHeader), "a client cannot forge the grant") require.Equal(t, "1", got.header.Get("X-Keep"), "other headers still pass") require.Empty(t, got.header.Get("Authorization")) } + +// a grant that lists endpoints opens only those: the first path segment under +// the deployment is the endpoint, and one outside the list is forbidden before +// any worker sees the request. A grant with no endpoints opens every one. +func TestFrontThreadGrantEndpointsScopeTheRoute(t *testing.T) { + ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")}) + scoped := mintToken(t, &auth.AgentThreadGrant{AgentName: "a", Sub: "demo-user", Send: true, Endpoints: []string{"front-desk"}}) + open := mintToken(t, &auth.AgentThreadGrant{AgentName: "a", Sub: "demo-user", Send: true}) + + post := func(tok string) int { + req, err := http.NewRequest(http.MethodPost, ts.URL+PathPrefix+"a/d/a2a", strings.NewReader("hello")) + require.NoError(t, err) + req.Header.Set("Authorization", "Bearer "+tok) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + resp.Body.Close() + return resp.StatusCode + } + require.Equal(t, http.StatusForbidden, post(scoped), "a2a is not among the grant's endpoints") + require.Equal(t, http.StatusNoContent, post(open)) + got := <-seen + require.Empty(t, got.pre.GetThreadGrant().GetEndpoints()) +} diff --git a/pkg/service/agentendpoint.go b/pkg/service/agentendpoint.go index 0b79d6127..50197fbce 100644 --- a/pkg/service/agentendpoint.go +++ b/pkg/service/agentendpoint.go @@ -17,6 +17,9 @@ package service import ( "net/http" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "github.com/livekit/livekit-server/pkg/agent/endpoint" ) @@ -38,9 +41,9 @@ func NewAgentEndpointService(h *AgentHandler, scopes *EndpointScopes) *AgentEndp level = endpoint.AccessGranted } return endpoint.Access{ - Scope: scopes.Scope(GetAPIKey(r.Context()), agentName, deployment), - Level: level, - SessionGrant: claims.AgentSession, + Scope: scopes.Scope(GetAPIKey(r.Context()), agentName, deployment), + Level: level, + ThreadGrant: claims.AgentThread, }, true } // unauthenticated: one configured key makes the api key @@ -61,6 +64,23 @@ func NewAgentEndpointService(h *AgentHandler, scopes *EndpointScopes) *AgentEndp Logger: h.logger, MaxTunnelsPerWorker: int(h.endpointsConfig.MaxTunnels), TunnelDrainTimeout: h.endpointsConfig.TunnelDrainTimeout, + ResolveThread: threadResolver(h), }), } } + +// threadResolver is the front's thread lookup when the server is configured +// with an AgentThreads address; nil otherwise, and a text route is forwarded +// unchanged. +func threadResolver(h *AgentHandler) endpoint.ThreadResolver { + address := h.endpointsConfig.ThreadsAddress + if address == "" { + return nil + } + conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + h.logger.Warnw("agent threads address is not dialable; threads are left to the workers", err, "address", address) + return nil + } + return endpoint.NewThreadResolver(conn, h.endpointsConfig.ThreadLookupTimeout, h.logger) +}