Add the agent HTTP endpoints data plane protocol

.changeset/agent-endpoint-grant.md

@@ -0,0 +1,6 @@
+---
+"github.com/livekit/protocol": patch
+"@livekit/protocol": patch
+---
+
+Add `AgentEndpointGrant`, scoping calls to an agent's non-public HTTP endpoints

agent/environment_test.go

@@ -0,0 +1,43 @@
+package agent
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestValidateDeployment(t *testing.T) {
+	cases := []struct {
+		name       string
+		deployment string
+		valid      bool
+	}{
+		{"empty", "", true},
+		{"alphanumeric", "production", true},
+		{"hyphen and dot", "prod-us.v2", true},
+		{"colon", "prod:us", true},
+		{"slash", "a/b", true},
+		{"non-ascii", "prodüction", true},
+		{"max length", strings.Repeat("a", MaxDeploymentLength), true},
+
+		{"underscore reserved", "prod_us", false},
+		{"space", "prod us", false},
+		{"tab", "prod\tus", false},
+		{"newline", "prod\nus", false},
+		{"del", "prod\x7fus", false},
+		{"nul", "prod\x00us", false},
+		{"too long", strings.Repeat("a", MaxDeploymentLength+1), false},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			err := ValidateDeployment(c.deployment)
+			if c.valid {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+			}
+		})
+	}
+}

auth/accesstoken.go

@@ -109,6 +109,11 @@ func (t *AccessToken) SetObservabilityGrant(grant *ObservabilityGrant) *AccessTo
 	return t
 }
 
+func (t *AccessToken) SetAgentEndpointGrant(grant *AgentEndpointGrant) *AccessToken {
+	t.grant.AgentEndpoint = grant
+	return t
+}
+
 func (t *AccessToken) SetMetadata(md string) *AccessToken {
 	t.grant.Metadata = md
 	return t

auth/accesstoken_test.go

@@ -223,3 +223,33 @@ func TestAccessToken(t *testing.T) {
 func apiKeypair() (string, string) {
 	return guid.New(utils.APIKeyPrefix), utils.RandomSecret()
 }
+
+func TestAgentEndpointGrantRoundTrip(t *testing.T) {
+	t.Parallel()
+
+	apiKey, secret := apiKeypair()
+	grant := &AgentEndpointGrant{Call: true, AgentName: "my-agent", Deployment: "prod"}
+	raw, err := NewAccessToken(apiKey, secret).
+		SetAgentEndpointGrant(grant).
+		SetValidFor(time.Minute).
+		ToJWT()
+	require.NoError(t, err)
+
+	// the claim key is camelCase
+	require.Contains(t, decodeClaims(t, raw), `"agentEndpoint"`)
+
+	v, err := ParseAPIToken(raw)
+	require.NoError(t, err)
+	_, decoded, err := v.Verify(secret)
+	require.NoError(t, err)
+	require.Equal(t, grant, decoded.AgentEndpoint)
+}
+
+func decodeClaims(t *testing.T, raw string) string {
+	t.Helper()
+	parts := strings.Split(raw, ".")
+	require.Len(t, parts, 3)
+	body, err := base64.RawURLEncoding.DecodeString(parts[1])
+	require.NoError(t, err)
+	return string(body)
+}

auth/grants.go

@@ -168,6 +168,7 @@ type ClaimGrants struct {
 	Agent         *AgentGrant         `json:"agent,omitempty"`
 	Inference     *InferenceGrant     `json:"inference,omitempty"`
 	Observability *ObservabilityGrant `json:"observability,omitempty"`
+	AgentEndpoint *AgentEndpointGrant `json:"agentEndpoint,omitempty"`
 	// Room configuration to use if this participant initiates the room
 	RoomConfig *RoomConfiguration `json:"roomConfig,omitempty"`
 	// Cloud-only, config preset to use
@@ -214,6 +215,7 @@ func (c *ClaimGrants) Clone() *ClaimGrants {
 	clone.Agent = c.Agent.Clone()
 	clone.Inference = c.Inference.Clone()
 	clone.Observability = c.Observability.Clone()
+	clone.AgentEndpoint = c.AgentEndpoint.Clone()
 	clone.Attributes = maps.Clone(c.Attributes)
 	clone.RoomConfig = c.RoomConfig.Clone()
 	if len(c.KindDetails) > 0 {
@@ -236,6 +238,7 @@ func (c *ClaimGrants) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	e.AddObject("Agent", c.Agent)
 	e.AddObject("Inference", c.Inference)
 	e.AddObject("Observability", c.Observability)
+	e.AddObject("AgentEndpoint", c.AgentEndpoint)
 	e.AddObject("RoomConfig", logger.Proto((*livekit.RoomConfiguration)(c.RoomConfig)))
 	e.AddString("RoomPreset", c.RoomPreset)
 	return nil
@@ -445,14 +448,14 @@ func (v *VideoGrant) UpdateFromPermission(permission *livekit.ParticipantPermiss
 
 func (v *VideoGrant) ToPermission() *livekit.ParticipantPermission {
 	return &livekit.ParticipantPermission{
-		CanPublish:          v.GetCanPublish(),
-		CanPublishData:      v.GetCanPublishData(),
-		CanSubscribe:        v.GetCanSubscribe(),
-		CanPublishSources:   v.GetCanPublishSources(),
-		CanUpdateMetadata:   v.GetCanUpdateOwnMetadata(),
-		Hidden:              v.Hidden,
-		Recorder:            v.Recorder,
-		Agent:               v.Agent,
+		CanPublish:            v.GetCanPublish(),
+		CanPublishData:        v.GetCanPublishData(),
+		CanSubscribe:          v.GetCanSubscribe(),
+		CanPublishSources:     v.GetCanPublishSources(),
+		CanUpdateMetadata:     v.GetCanUpdateOwnMetadata(),
+		Hidden:                v.Hidden,
+		Recorder:              v.Recorder,
+		Agent:                 v.Agent,
 		CanSubscribeMetrics:   v.GetCanSubscribeMetrics(),
 		CanManageAgentSession: v.GetCanManageAgentSession(),
 	}
@@ -663,6 +666,52 @@ func (s *ObservabilityGrant) MarshalLogObject(e zapcore.ObjectEncoder) error {
 
 // ------------------------------------------------------------------
 
+type AgentEndpointGrant struct {
+	// Call grants to invoke an agent's non-public HTTP endpoints.
+	Call bool `json:"call,omitempty"`
+	// AgentName restricts the grant to one agent; empty grants every agent in the project.
+	AgentName string `json:"agentName,omitempty"`
+	// Deployment restricts the grant to one deployment; empty grants every
+	// deployment. A worker registered without one is addressed as "default".
+	Deployment string `json:"deployment,omitempty"`
+}
+
+// Allows reports whether the grant authorizes calling non-public endpoints of
+// (agentName, deployment). Matching is exact and case-sensitive; an empty scope
+// field matches any value.
+func (s *AgentEndpointGrant) Allows(agentName, deployment string) bool {
+	if s == nil || !s.Call {
+		return false
+	}
+	if s.AgentName != "" && s.AgentName != agentName {
+		return false
+	}
+	return s.Deployment == "" || s.Deployment == deployment
+}
+
+func (s *AgentEndpointGrant) Clone() *AgentEndpointGrant {
+	if s == nil {
+		return nil
+	}
+
+	clone := *s
+
+	return &clone
+}
+
+func (s *AgentEndpointGrant) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	if s == nil {
+		return nil
+	}
+
+	e.AddBool("Call", s.Call)
+	e.AddString("AgentName", s.AgentName)
+	e.AddString("Deployment", s.Deployment)
+	return nil
+}
+
+// ------------------------------------------------------------------
+
 func sourceToString(source livekit.TrackSource) string {
 	return strings.ToLower(source.String())
 }

auth/grants_test.go

@@ -35,6 +35,7 @@ func TestGrants(t *testing.T) {
 		require.Same(t, grants.Agent, clone.Agent)
 		require.Same(t, grants.Inference, clone.Inference)
 		require.Same(t, grants.SIP, clone.SIP)
+		require.Same(t, grants.AgentEndpoint, clone.AgentEndpoint)
 		require.True(t, reflect.DeepEqual(grants, clone))
 		require.True(t, reflect.DeepEqual(grants.Video, clone.Video))
 	})
@@ -61,6 +62,9 @@ func TestGrants(t *testing.T) {
 		// require Inference
 		require.Same(t, grants.Inference, clone.Inference)
 		require.True(t, reflect.DeepEqual(grants.Inference, clone.Inference))
+		// require AgentEndpoint
+		require.Same(t, grants.AgentEndpoint, clone.AgentEndpoint)
+		require.True(t, reflect.DeepEqual(grants.AgentEndpoint, clone.AgentEndpoint))
 	})
 
 	t.Run("clone with video", func(t *testing.T) {
@@ -475,3 +479,48 @@ func TestRoomConfiguration_CheckCredentials(t *testing.T) {
 		require.NoError(t, config.CheckCredentials())
 	})
 }
+
+func TestAgentEndpointGrantAllows(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name       string
+		grant      *AgentEndpointGrant
+		agentName  string
+		deployment string
+		want       bool
+	}{
+		{"nil grant", nil, "a", "prod", false},
+		{"call unset with scope", &AgentEndpointGrant{AgentName: "a", Deployment: "prod"}, "a", "prod", false},
+		{"wildcard", &AgentEndpointGrant{Call: true}, "a", "prod", true},
+		{"agent match", &AgentEndpointGrant{Call: true, AgentName: "a"}, "a", "prod", true},
+		{"agent mismatch", &AgentEndpointGrant{Call: true, AgentName: "a"}, "b", "prod", false},
+		{"deployment match", &AgentEndpointGrant{Call: true, Deployment: "prod"}, "a", "prod", true},
+		{"deployment mismatch", &AgentEndpointGrant{Call: true, Deployment: "prod"}, "a", "staging", false},
+		{"both match", &AgentEndpointGrant{Call: true, AgentName: "a", Deployment: "prod"}, "a", "prod", true},
+		{"both set, deployment differs", &AgentEndpointGrant{Call: true, AgentName: "a", Deployment: "prod"}, "a", "staging", false},
+		{"agent case differs", &AgentEndpointGrant{Call: true, AgentName: "Agent"}, "agent", "prod", false},
+		{"deployment case differs", &AgentEndpointGrant{Call: true, Deployment: "Prod"}, "a", "prod", false},
+		{"empty deployment is wildcard", &AgentEndpointGrant{Call: true}, "a", "production", true},
+		{"default pinned by literal", &AgentEndpointGrant{Call: true, Deployment: "default"}, "a", "default", true},
+		{"default pin rejects others", &AgentEndpointGrant{Call: true, Deployment: "default"}, "a", "prod", false},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			require.Equal(t, c.want, c.grant.Allows(c.agentName, c.deployment))
+		})
+	}
+}
+
+func TestAgentEndpointGrantCloneIndependent(t *testing.T) {
+	t.Parallel()
+
+	grants := &ClaimGrants{AgentEndpoint: &AgentEndpointGrant{Call: true, AgentName: "a", Deployment: "prod"}}
+	clone := grants.Clone()
+	require.NotSame(t, grants.AgentEndpoint, clone.AgentEndpoint)
+
+	clone.AgentEndpoint.Call = false
+	clone.AgentEndpoint.AgentName = "b"
+	require.True(t, grants.AgentEndpoint.Call)
+	require.Equal(t, "a", grants.AgentEndpoint.AgentName)
+}