diff --git a/pkg/agent/endpoint/attempt.go b/pkg/agent/endpoint/attempt.go index 10e1e854b..57e3e34db 100644 --- a/pkg/agent/endpoint/attempt.go +++ b/pkg/agent/endpoint/attempt.go @@ -34,6 +34,11 @@ import ( "github.com/livekit/protocol/utils/guid" ) +// sessionGrantHeader 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" + // attempt is the per-request state shared across worker attempts. type attempt struct { req *http.Request @@ -43,6 +48,9 @@ type attempt struct { target string requestID string granted bool + // grant is the caller's verified agent session 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 @@ -99,10 +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, - ClientAddr: clientAddr, - Scheme: scheme, + RequestId: a.requestID, + Authorized: a.granted, + SessionGrant: a.grant, + ClientAddr: clientAddr, + Scheme: scheme, } return p } @@ -132,11 +141,15 @@ func (a *attempt) requestHeader() http.Header { h.Set("Connection", "Upgrade") h.Set("Upgrade", "websocket") } - // identity travels as preamble.authorized; the token must never reach the - // worker + // identity travels as preamble.authorized and preamble.session_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 + // 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) // 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 29a4b5482..770e1fa61 100644 --- a/pkg/agent/endpoint/front.go +++ b/pkg/agent/endpoint/front.go @@ -32,6 +32,7 @@ import ( "github.com/livekit/livekit-server/pkg/agent/endpoint/router" "github.com/livekit/livekit-server/pkg/agent/endpoint/wire" + "github.com/livekit/protocol/auth" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" ) @@ -118,6 +119,10 @@ 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 + // token carried none or there was no token. It reaches the worker in the + // preamble; the token itself never does. + SessionGrant *auth.AgentSessionGrant } // AccessResolver maps an inbound request, plus the agent and deployment its URL @@ -295,6 +300,7 @@ 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, webSocket: isWebSocketUpgrade(r), pools: f.pools, diff --git a/pkg/agent/endpoint/front_test.go b/pkg/agent/endpoint/front_test.go index 47ed7eed1..adf362100 100644 --- a/pkg/agent/endpoint/front_test.go +++ b/pkg/agent/endpoint/front_test.go @@ -15,6 +15,7 @@ import ( "github.com/stretchr/testify/require" + "github.com/livekit/protocol/auth" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" @@ -414,12 +415,15 @@ func TestMergeWorkerHeaders(t *testing.T) { // textExchange is what the scripted worker saw of one request. type textExchange struct { - pre *livekit.AgentHttp_StreamPreamble - body string + pre *livekit.AgentHttp_StreamPreamble + target string + header http.Header + body string } // textFront serves one scripted worker declaring a text-kind route next to a -// plain one, and reports each exchange it served. +// plain one, and reports each exchange it served. A nil ResolveAccess grants +// every request. func textFront(t *testing.T, params FrontParams) (*httptest.Server, <-chan textExchange) { t.Helper() m, err := ParseManifest([]*livekit.AgentHttp_AgentEndpoint{ @@ -442,11 +446,15 @@ func textFront(t *testing.T, params FrontParams) (*httptest.Server, <-chan textE return } body, _ := io.ReadAll(req.Body) - seen <- textExchange{pre: pre, body: string(body)} + seen <- textExchange{pre: pre, target: req.RequestURI, header: req.Header, body: string(body)} w.WriteString("HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n") }, }})) - params.ResolveAccess = grantedTo(s) + if params.ResolveAccess == nil { + params.ResolveAccess = grantedTo(s) + } else { + params.ResolveAccess = withScope(params.ResolveAccess, s) + } params.Logger = logger.GetLogger() ts := httptest.NewServer(NewFront(params)) t.Cleanup(ts.Close) @@ -554,3 +562,133 @@ func TestFrontSessionPeekShortBody(t *testing.T) { require.Equal(t, "AT_2", got.pre.GetSessionId()) require.Empty(t, got.body) } + +const testAPIKey, testAPISecret = "APIfront", "front-secret-front-secret-front-secret" + +// 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. +func verifiedAccess(agentName, deployment string) AccessResolver { + return func(r *http.Request, _, _ string) (Access, bool) { + raw := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") + if r.Header.Get("Authorization") == "" { + raw = r.URL.Query().Get("access_token") + } + if raw == "" { + return Access{}, true + } + v, err := auth.ParseAPIToken(raw) + if err != nil { + return Access{}, false + } + _, claims, err := v.Verify(testAPISecret) + if err != nil { + return Access{}, false + } + level := AccessCredentialed + if claims.AgentEndpoint.Allows(agentName, deployment) { + level = AccessGranted + } + return Access{Level: level, SessionGrant: claims.AgentSession}, true + } +} + +// withScope places whatever resolve decides in scope. +func withScope(resolve AccessResolver, scope *Scope) AccessResolver { + return func(r *http.Request, agentName, deployment string) (Access, bool) { + a, ok := resolve(r, agentName, deployment) + a.Scope = scope + return a, ok + } +} + +func mintToken(t *testing.T, grant *auth.AgentSessionGrant) string { + t.Helper() + tok, err := auth.NewAccessToken(testAPIKey, testAPISecret). + SetIdentity("caller"). + SetAgentEndpointGrant(&auth.AgentEndpointGrant{Call: true}). + SetAgentSessionGrant(grant). + ToJWT() + require.NoError(t, err) + return tok +} + +// the verified token's agentSession 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) { + ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")}) + grant := &auth.AgentSessionGrant{ + AgentName: "a", Sub: "demo-user", SessionIDs: []string{"AT_1", "AT_2"}, + List: true, Create: true, Send: true, Read: true, + } + tok := mintToken(t, grant) + + for _, path := range []string{"/a2a", "/plain"} { + req, err := http.NewRequest(http.MethodPost, ts.URL+PathPrefix+"a/d"+path, 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() + require.Equal(t, http.StatusNoContent, resp.StatusCode) + + 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()) + } +} + +// 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) { + ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")}) + tok := mintToken(t, &auth.AgentSessionGrant{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) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + resp.Body.Close() + 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.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) { + 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("X-Keep", "1") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + resp.Body.Close() + require.Equal(t, http.StatusNoContent, resp.StatusCode) + + 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.Equal(t, "1", got.header.Get("X-Keep"), "other headers still pass") + require.Empty(t, got.header.Get("Authorization")) +} diff --git a/pkg/agent/endpoint/wire/exchange.go b/pkg/agent/endpoint/wire/exchange.go index b6299fe5d..3614d8c86 100644 --- a/pkg/agent/endpoint/wire/exchange.go +++ b/pkg/agent/endpoint/wire/exchange.go @@ -142,7 +142,9 @@ func ReadPreamble(r io.Reader) (*livekit.AgentHttp_StreamPreamble, error) { // StripReservedHeaders removes the in-band signalling fields. Every header set // crossing between client and worker must pass through here: a client-set // x-lk-completion would otherwise forge an outcome, and a worker-set one would -// reach the end client. +// reach the end client. Identity headers (the LiveKit bearer, the +// X-LiveKit-Session-Grant a node-less worker accepts) are the front's to strip +// on the request side, next to where it asserts them in the preamble. func StripReservedHeaders(h http.Header) { for k := range h { if strings.HasPrefix(strings.ToLower(k), ReservedHeaderPrefix) { diff --git a/pkg/service/agentendpoint.go b/pkg/service/agentendpoint.go index ec6761a0b..0b79d6127 100644 --- a/pkg/service/agentendpoint.go +++ b/pkg/service/agentendpoint.go @@ -38,8 +38,9 @@ func NewAgentEndpointService(h *AgentHandler, scopes *EndpointScopes) *AgentEndp level = endpoint.AccessGranted } return endpoint.Access{ - Scope: scopes.Scope(GetAPIKey(r.Context()), agentName, deployment), - Level: level, + Scope: scopes.Scope(GetAPIKey(r.Context()), agentName, deployment), + Level: level, + SessionGrant: claims.AgentSession, }, true } // unauthenticated: one configured key makes the api key