diff --git a/pkg/agent/config.go b/pkg/agent/config.go index d8e82ee9d..ffcc26ed1 100644 --- a/pkg/agent/config.go +++ b/pkg/agent/config.go @@ -17,15 +17,6 @@ type EndpointsConfig struct { // rejects registrations that declare endpoints. Disabled bool `yaml:"disabled,omitempty"` - // WebTransportPort is the UDP port for the worker WebTransport listener that - // carries the control stream and HTTP exchanges. 0 disables the listener, so - // no worker can serve endpoints (the front then only relays, in cloud). - WebTransportPort uint32 `yaml:"webtransport_port,omitempty"` - // TLSCertFile / TLSKeyFile is the certificate the WebTransport (QUIC) listener - // presents. QUIC has no plaintext mode, so a listener needs one; in dev mode a - // self-signed cert is generated when these are unset. - TLSCertFile string `yaml:"tls_cert_file,omitempty"` - TLSKeyFile string `yaml:"tls_key_file,omitempty"` // MaxStreams is the soft per-session concurrent-stream cap used for capacity // weighting; 0 takes the endpoint package default. MaxStreams uint32 `yaml:"max_streams,omitempty"` diff --git a/pkg/agent/endpoint/manifest.go b/pkg/agent/endpoint/manifest.go index e781f2429..fde4e17e4 100644 --- a/pkg/agent/endpoint/manifest.go +++ b/pkg/agent/endpoint/manifest.go @@ -15,12 +15,15 @@ package endpoint import ( + "errors" "fmt" "net/http" "slices" "strings" "github.com/livekit/protocol/livekit" + + "github.com/livekit/livekit-server/pkg/agent/endpoint/wire" ) const MaxManifestRoutes = 256 @@ -64,6 +67,30 @@ const ( ) // ParseManifest validates a registration's endpoint list. +// NegotiateSettings validates a registration's endpoint manifest and negotiates +// the data-plane protocol version. The data plane is a WebTransport session (no +// attach token, no fixed connection pool), so the version is all there is to +// agree on. +func NegotiateSettings(req *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) { + if _, err := ParseManifest(req.GetEndpoints()); err != nil { + return nil, err + } + if req.GetInstanceId() == "" { + return nil, errors.New("registrations with endpoints require an instance_id") + } + if req.GetEndpointProtocol() == 0 { + return nil, errors.New("worker declared endpoints but no endpoint protocol; upgrade the agent SDK to one that speaks the endpoint data plane") + } + // the worker frames to the version returned here, so it must be the + // negotiated one and not the server's own constant + negotiated := min(req.GetEndpointProtocol(), wire.CurrentProtocol) + if negotiated < wire.MinProtocol { + return nil, fmt.Errorf("unsupported agent endpoint protocol %d (this server serves %d..%d)", + req.GetEndpointProtocol(), wire.MinProtocol, wire.CurrentProtocol) + } + return &livekit.AgentHttp_AgentEndpointSettings{Protocol: negotiated}, nil +} + func ParseManifest(endpoints []*livekit.AgentHttp_AgentEndpoint) (*Manifest, error) { if len(endpoints) > MaxManifestRoutes { return nil, fmt.Errorf("manifest exceeds %d routes", MaxManifestRoutes) diff --git a/pkg/agent/testutils/server.go b/pkg/agent/testutils/server.go index 3f490918c..3bb7934a2 100644 --- a/pkg/agent/testutils/server.go +++ b/pkg/agent/testutils/server.go @@ -15,6 +15,7 @@ import ( "github.com/gammazero/deque" "github.com/livekit/livekit-server/pkg/agent" + "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/pkg/service" @@ -40,7 +41,7 @@ type TestServer struct { func NewTestServer(bus psrpc.MessageBus) *TestServer { localNode, _ := routing.NewLocalNode(nil) - return NewTestServerWithService(must.Get(service.NewAgentService( + return NewTestServerWithService(must.Get(service.NewAgentHandler( &config.Config{ Region: "test", Agents: agent.Config{ @@ -50,6 +51,7 @@ func NewTestServer(bus psrpc.MessageBus) *TestServer { localNode, bus, auth.NewSimpleKeyProvider("test", "verysecretsecret"), + endpoint.NewRegistry(), ))) } diff --git a/pkg/agent/worker.go b/pkg/agent/worker.go index ca913e9eb..413425d55 100644 --- a/pkg/agent/worker.go +++ b/pkg/agent/worker.go @@ -179,29 +179,52 @@ func MakeWorkerRegistration() WorkerRegistration { var _ WorkerSignalHandler = (*WorkerRegisterer)(nil) -// EndpointSettingsFunc validates a registration's endpoint manifest and returns the -// negotiated data-plane settings (including the attach token). Returning an error -// fails the registration. It runs only when the registration declares endpoints AND -// a data-plane protocol version; a nil func rejects such registrations. -type EndpointSettingsFunc func(req *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) +// WorkerRegisterHandler takes a turn during a worker's registration: it reads the +// request, validates the part it owns, and fills in its part of the response and +// the registration. Returning an error fails the registration. +type WorkerRegisterHandler func(req *livekit.RegisterWorkerRequest, res *livekit.RegisterWorkerResponse, reg *WorkerRegistration) error + +// EndpointRegisterHandler owns a worker's endpoint declaration: it validates the +// declaration and negotiates the data-plane settings. A registration declaring no +// endpoints passes through untouched. +func EndpointRegisterHandler(req *livekit.RegisterWorkerRequest, res *livekit.RegisterWorkerResponse, reg *WorkerRegistration) error { + if len(req.GetEndpoints()) == 0 { + return nil + } + // 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 endpoint.IsReservedAgentName(req.GetAgentName()) { + return fmt.Errorf("agent name %q is reserved and cannot serve HTTP endpoints", req.GetAgentName()) + } + settings, err := endpoint.NegotiateSettings(req) + if err != nil { + return err + } + res.EndpointSettings = settings + reg.EndpointSettings = settings + return nil +} type WorkerRegisterer struct { WorkerPingHandler - serverInfo *livekit.ServerInfo - deadline time.Time - endpointSettings EndpointSettingsFunc + serverInfo *livekit.ServerInfo + deadline time.Time + handlers []WorkerRegisterHandler registration WorkerRegistration registered bool } -func NewWorkerRegisterer(conn SignalConn, serverInfo *livekit.ServerInfo, base WorkerRegistration, endpointSettings EndpointSettingsFunc) *WorkerRegisterer { +func NewWorkerRegisterer(conn SignalConn, serverInfo *livekit.ServerInfo, base WorkerRegistration, handlers ...WorkerRegisterHandler) *WorkerRegisterer { return &WorkerRegisterer{ WorkerPingHandler: WorkerPingHandler{conn: conn}, serverInfo: serverInfo, registration: base, deadline: time.Now().Add(RegisterTimeout), - endpointSettings: endpointSettings, + handlers: handlers, } } @@ -245,34 +268,19 @@ func (h *WorkerRegisterer) HandleRegister(req *livekit.RegisterWorkerRequest) er h.registration.Endpoints = req.GetEndpoints() h.registration.InstanceID = req.GetInstanceId() - if len(req.GetEndpoints()) > 0 { - if h.endpointSettings == nil { - return errors.New("agent HTTP endpoints are not supported by this server") - } - // 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 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 { + res := &livekit.RegisterWorkerResponse{ + WorkerId: h.registration.ID, + ServerInfo: h.serverInfo, + } + for _, handle := range h.handlers { + if err := handle(req, res, &h.registration); err != nil { return err } - h.registration.EndpointSettings = settings } h.registered = true _, err := h.conn.WriteServerMessage(&livekit.ServerMessage{ - Message: &livekit.ServerMessage_Register{ - Register: &livekit.RegisterWorkerResponse{ - WorkerId: h.registration.ID, - ServerInfo: h.serverInfo, - EndpointSettings: h.registration.EndpointSettings, - }, - }, + Message: &livekit.ServerMessage_Register{Register: res}, }) return err } diff --git a/pkg/agent/worker_test.go b/pkg/agent/worker_test.go index b24cc82e3..5b9947892 100644 --- a/pkg/agent/worker_test.go +++ b/pkg/agent/worker_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/require" "github.com/livekit/livekit-server/pkg/agent" + "github.com/livekit/livekit-server/pkg/agent/endpoint/wire" "github.com/livekit/protocol/livekit" ) @@ -47,14 +48,14 @@ func TestHandleRegisterEndpointAgentNames(t *testing.T) { for _, c := range cases { t.Run(c.name, func(t *testing.T) { h := agent.NewWorkerRegisterer(nopSignalConn{}, &livekit.ServerInfo{}, agent.WorkerRegistration{}, - func(*livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) { - return &livekit.AgentHttp_AgentEndpointSettings{}, nil - }) + agent.EndpointRegisterHandler) err := h.HandleRegister(&livekit.RegisterWorkerRequest{ - Type: livekit.JobType_JT_ROOM, - AgentName: c.agentName, - Endpoints: []*livekit.AgentHttp_AgentEndpoint{{Path: "/hook", Methods: []string{"GET"}}}, + Type: livekit.JobType_JT_ROOM, + AgentName: c.agentName, + Endpoints: []*livekit.AgentHttp_AgentEndpoint{{Path: "/hook", Methods: []string{"GET"}}}, + InstanceId: "AEI_test", + EndpointProtocol: wire.CurrentProtocol, }) if c.ok { require.NoError(t, err) @@ -68,7 +69,7 @@ func TestHandleRegisterEndpointAgentNames(t *testing.T) { // 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{}, nil) + h := agent.NewWorkerRegisterer(nopSignalConn{}, &livekit.ServerInfo{}, agent.WorkerRegistration{}, agent.EndpointRegisterHandler) require.NoError(t, h.HandleRegister(&livekit.RegisterWorkerRequest{ Type: livekit.JobType_JT_ROOM, AgentName: name, diff --git a/pkg/config/config.go b/pkg/config/config.go index 3ca89107e..bc03d5705 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -82,6 +82,7 @@ type Config struct { PrometheusPort uint32 `yaml:"prometheus_port,omitempty"` Prometheus PrometheusConfig `yaml:"prometheus,omitempty"` DebugHandler DebugHandlerConfig `yaml:"debug_handler,omitempty"` + WebTransport WebTransportConfig `yaml:"webtransport,omitempty"` RTC RTCConfig `yaml:"rtc,omitempty"` Redis redisLiveKit.RedisConfig `yaml:"redis,omitempty"` Audio sfu.AudioConfig `yaml:"audio,omitempty"` @@ -117,6 +118,15 @@ type Config struct { API APIConfig `yaml:"api,omitempty"` } +// WebTransportConfig is the node's HTTP/3 listener. QUIC has no plaintext mode, +// so a certificate is required; development mode generates a self-signed one. +type WebTransportConfig struct { + // Port is the UDP port to listen on. 0 starts no listener. + Port uint32 `yaml:"port,omitempty"` + TLSCertFile string `yaml:"tls_cert_file,omitempty"` + TLSKeyFile string `yaml:"tls_key_file,omitempty"` +} + type RTCConfig struct { rtcconfig.RTCConfig `yaml:",inline"` diff --git a/pkg/service/agentendpoint.go b/pkg/service/agentendpoint.go new file mode 100644 index 000000000..b05271b8d --- /dev/null +++ b/pkg/service/agentendpoint.go @@ -0,0 +1,50 @@ +// 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 ( + "net/http" + + "github.com/livekit/livekit-server/pkg/agent/endpoint" +) + +// AgentEndpointService is the /agents/{agent_name}/{deployment}/{path...} front +// backed by this node's attached workers. The api key comes from validated grants +// when a token is present; a non-public route additionally requires an +// agent-endpoint grant scoped to this agent and deployment. +type AgentEndpointService struct { + *endpoint.Front +} + +func NewAgentEndpointService(h *AgentHandler, registry *endpoint.Registry) *AgentEndpointService { + return &AgentEndpointService{ + Front: endpoint.NewFront(endpoint.FrontParams{ + Registry: registry, + ResolveAccess: func(r *http.Request, agentName, deployment string) endpoint.Access { + if claims := GetGrants(r.Context()); claims != nil { + level := endpoint.AccessCredentialed + if claims.AgentEndpoint.Allows(agentName, deployment) { + level = endpoint.AccessGranted + } + return endpoint.Access{APIKey: GetAPIKey(r.Context()), Level: level} + } + // unauthenticated: one configured key makes the api key unambiguous + return endpoint.Access{APIKey: h.singleAPIKey} + }, + Logger: h.logger, + SingleKeyFallback: true, + }), + } +} diff --git a/pkg/service/agentendpoint_protocol_test.go b/pkg/service/agentendpoint_protocol_test.go index 542cc2b23..40ac8439b 100644 --- a/pkg/service/agentendpoint_protocol_test.go +++ b/pkg/service/agentendpoint_protocol_test.go @@ -35,29 +35,46 @@ func TestEndpointSettingsNegotiation(t *testing.T) { } } h := &AgentHandler{} + negotiate := func(h *AgentHandler, r *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) { + res := &livekit.RegisterWorkerResponse{} + reg := &agent.WorkerRegistration{} + if err := h.endpointRegisterHandler(r, res, reg); err != nil { + return nil, err + } + require.Equal(t, res.GetEndpointSettings(), reg.EndpointSettings, + "the registration must carry what the worker was told") + return res.GetEndpointSettings(), nil + } // a worker predating the data plane declares endpoints with protocol 0: it // cannot frame at all, and the error must name that rather than read as a // version out of range - _, err := h.endpointSettings(req(0)) + _, err := negotiate(h, req(0)) require.ErrorContains(t, err, "no endpoint protocol") - settings, err := h.endpointSettings(req(wire.CurrentProtocol)) + settings, err := negotiate(h, req(wire.CurrentProtocol)) require.NoError(t, err) require.Equal(t, wire.CurrentProtocol, settings.GetProtocol()) // a worker speaking a newer version is negotiated down to what this server // serves; the worker frames to the version returned here - settings, err = h.endpointSettings(req(wire.CurrentProtocol + 42)) + settings, err = negotiate(h, req(wire.CurrentProtocol+42)) require.NoError(t, err) require.Equal(t, wire.CurrentProtocol, settings.GetProtocol()) noInstance := req(wire.CurrentProtocol) noInstance.InstanceId = "" - _, err = h.endpointSettings(noInstance) + _, err = negotiate(h, noInstance) require.ErrorContains(t, err, "instance_id") disabled := &AgentHandler{endpointsConfig: agent.EndpointsConfig{Disabled: true}} - _, err = disabled.endpointSettings(req(wire.CurrentProtocol)) + _, err = negotiate(disabled, req(wire.CurrentProtocol)) require.ErrorContains(t, err, "disabled") + + // a registration declaring no endpoints is left alone, even with endpoints off + bare := req(wire.CurrentProtocol) + bare.Endpoints = nil + settings, err = negotiate(disabled, bare) + require.NoError(t, err) + require.Nil(t, settings) } diff --git a/pkg/service/agentendpoint_test.go b/pkg/service/agentendpoint_test.go index 9c1323813..a9ebdcda8 100644 --- a/pkg/service/agentendpoint_test.go +++ b/pkg/service/agentendpoint_test.go @@ -41,6 +41,7 @@ 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" @@ -59,10 +60,10 @@ const ( const testMaxAPIBodySize = 64 << 10 type endpointStack struct { - t *testing.T - ts *httptest.Server - svc *service.AgentService - wtURL string // https://host:port/agent (WebTransport control+data) + t *testing.T + ts *httptest.Server + handler *service.AgentHandler + wtURL string // https://host:port/agent (WebTransport control+data) } // selfSignedTLS mints an in-memory cert for 127.0.0.1 with the h3 ALPN, for the @@ -97,7 +98,8 @@ func newEndpointStack(t *testing.T, endpointsCfg agent.EndpointsConfig) *endpoin } conf.Limit.MaxAPIRequestBodySize = testMaxAPIBodySize - svc, err := service.NewAgentService(conf, localNode, psrpc.NewLocalMessageBus(), keyProvider) + registry := endpoint.NewRegistry() + h, err := service.NewAgentHandler(conf, localNode, psrpc.NewLocalMessageBus(), keyProvider, registry) require.NoError(t, err) // the production handler, so these tests run on the node's real middleware chain @@ -110,21 +112,23 @@ func newEndpointStack(t *testing.T, endpointsCfg agent.EndpointsConfig) *endpoin var agentFront http.Handler if !endpointsCfg.Disabled { - agentFront = svc.EndpointFront() + agentFront = service.NewAgentEndpointService(h, registry) } ts := httptest.NewServer(service.NewHTTPHandler(conf, keyProvider, apiMux, agentFront)) t.Cleanup(ts.Close) - t.Cleanup(func() { svc.DrainConnections(time.Millisecond, true) }) + t.Cleanup(func() { h.DrainConnections(time.Millisecond, true) }) // workers connect over WebTransport (control + data on one session) - udp, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + wtMux := http.NewServeMux() + wtMux.Handle("/agent", service.NewAgentWTService(h)) + wt := service.NewWebTransportServer(selfSignedTLS(t)) + wt.H3.Handler = service.NewWebTransportHandler(keyProvider, wt, wtMux) + bound, stopWT, err := service.ListenWebTransport(wt, []string{"127.0.0.1"}, 0) require.NoError(t, err) - wt := service.NewAgentWebTransportServer(svc, keyProvider, selfSignedTLS(t)) - go func() { _ = wt.Serve(udp) }() - t.Cleanup(func() { _ = wt.Close(); _ = udp.Close() }) - wtURL := "https://" + udp.LocalAddr().String() + "/agent" + t.Cleanup(stopWT) + wtURL := "https://" + bound[0].String() + "/agent" - return &endpointStack{t: t, ts: ts, svc: svc, wtURL: wtURL} + return &endpointStack{t: t, ts: ts, handler: h, wtURL: wtURL} } func (s *endpointStack) startWorker(target string, deployment string, endpoints []*livekit.AgentHttp_AgentEndpoint) *conformance.Worker { diff --git a/pkg/service/agentservice.go b/pkg/service/agenthandler.go similarity index 70% rename from pkg/service/agentservice.go rename to pkg/service/agenthandler.go index 6724ba8c5..641de68d6 100644 --- a/pkg/service/agentservice.go +++ b/pkg/service/agenthandler.go @@ -17,85 +17,29 @@ package service import ( "context" "errors" - "fmt" "math/rand" - "net/http" "slices" "sort" - "strconv" "sync" "time" - "github.com/gorilla/websocket" "google.golang.org/protobuf/types/known/emptypb" - "github.com/livekit/livekit-server/pkg/agent" - "github.com/livekit/livekit-server/pkg/agent/endpoint" - "github.com/livekit/livekit-server/pkg/agent/endpoint/wire" - "github.com/livekit/livekit-server/pkg/config" - "github.com/livekit/livekit-server/pkg/routing" - "github.com/livekit/livekit-server/pkg/rtc" - "github.com/livekit/livekit-server/pkg/rtc/types" - "github.com/livekit/livekit-server/version" "github.com/livekit/protocol/auth" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/rpc" "github.com/livekit/protocol/utils" "github.com/livekit/psrpc" + + "github.com/livekit/livekit-server/pkg/agent" + "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/pkg/rtc/types" + "github.com/livekit/livekit-server/version" ) -type AgentSocketUpgrader struct { - websocket.Upgrader -} - -func (u AgentSocketUpgrader) Upgrade( - w http.ResponseWriter, - r *http.Request, - responseHeader http.Header, -) ( - conn *websocket.Conn, - registration agent.WorkerRegistration, - ok bool, -) { - if u.CheckOrigin == nil { - // allow connections from any origin, since script may be hosted anywhere - // security is enforced by access tokens - u.CheckOrigin = func(r *http.Request) bool { - return true - } - } - - // reject non websocket requests - if !websocket.IsWebSocketUpgrade(r) { - w.WriteHeader(404) - return - } - - // require a claim - claims := GetGrants(r.Context()) - if claims == nil || claims.Video == nil || !claims.Video.Agent { - HandleError(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied) - return - } - - registration = agent.MakeWorkerRegistration() - registration.ClientIP = GetClientIP(r) - - // upgrade - conn, err := u.Upgrader.Upgrade(w, r, responseHeader) - if err != nil { - HandleError(w, r, http.StatusInternalServerError, err) - return - } - - if pv, err := strconv.Atoi(r.FormValue("protocol")); err == nil { - registration.Protocol = agent.WorkerProtocolVersion(pv) - } - - return conn, registration, true -} - func DispatchAgentWorkerSignal(c agent.SignalConn, h agent.WorkerSignalHandler, l logger.Logger) bool { req, _, err := c.ReadWorkerMessage() if err != nil { @@ -115,8 +59,8 @@ func DispatchAgentWorkerSignal(c agent.SignalConn, h agent.WorkerSignalHandler, return true } -func HandshakeAgentWorker(c agent.SignalConn, serverInfo *livekit.ServerInfo, registration agent.WorkerRegistration, l logger.Logger, endpointSettings agent.EndpointSettingsFunc) (r agent.WorkerRegistration, ok bool) { - wr := agent.NewWorkerRegisterer(c, serverInfo, registration, endpointSettings) +func HandshakeAgentWorker(c agent.SignalConn, serverInfo *livekit.ServerInfo, registration agent.WorkerRegistration, l logger.Logger, handlers ...agent.WorkerRegisterHandler) (r agent.WorkerRegistration, ok bool) { + wr := agent.NewWorkerRegisterer(c, serverInfo, registration, handlers...) if err := c.SetReadDeadline(wr.Deadline()); err != nil { return } @@ -131,17 +75,6 @@ func HandshakeAgentWorker(c agent.SignalConn, serverInfo *livekit.ServerInfo, re return wr.Registration(), true } -type AgentService struct { - upgrader AgentSocketUpgrader - - signalMessageSizeLimit int64 - // developmentMode allows the WebTransport listener to run without a - // configured certificate. - developmentMode bool - - *AgentHandler -} - type AgentHandler struct { agentServer rpc.AgentInternalServer // the server's only configured api key, when exactly one exists: the @@ -179,138 +112,58 @@ type workerKey struct { deployment string } -func NewAgentService( +// NewAgentHandler builds the node's agent worker handler, shared by the +// transport services. +func NewAgentHandler( conf *config.Config, currentNode routing.LocalNode, bus psrpc.MessageBus, keyProvider auth.KeyProvider, -) (*AgentService, error) { - s := &AgentService{ - signalMessageSizeLimit: conf.Limit.AgentSignalMessageSizeLimit, - developmentMode: conf.Development, - } - - serverInfo := &livekit.ServerInfo{ - Edition: livekit.ServerInfo_Standard, - Version: version.Version, - Protocol: types.CurrentProtocol, - AgentProtocol: agent.CurrentProtocol, - Region: conf.Region, - NodeId: string(currentNode.NodeID()), - } - - agentServer, err := rpc.NewAgentInternalServer(s, bus) - if err != nil { - return nil, err - } - s.AgentHandler = NewAgentHandler( - agentServer, - keyProvider, - logger.GetLogger(), - serverInfo, - conf.Agents.TargetLoad, - agent.RoomAgentTopic, - agent.PublisherAgentTopic, - agent.ParticipantAgentTopic, - ) - s.AgentHandler.endpointsConfig = conf.Agents.Endpoints - if len(conf.Keys) == 1 { - for key := range conf.Keys { - s.AgentHandler.singleAPIKey = key - } - } - - return s, nil -} - -// EndpointFront is the /agents/{agent_name}/{deployment}/{path...} handler backed -// by this node's attached workers. The api key comes from validated grants when a -// token is present; a non-public route additionally requires an agent-endpoint -// grant scoped to this agent and deployment. -func (s *AgentService) EndpointFront() http.Handler { - return endpoint.NewFront(endpoint.FrontParams{ - Registry: s.endpointRegistry, - ResolveAccess: func(r *http.Request, agentName, deployment string) endpoint.Access { - if claims := GetGrants(r.Context()); claims != nil { - level := endpoint.AccessCredentialed - if claims.AgentEndpoint.Allows(agentName, deployment) { - level = endpoint.AccessGranted - } - return endpoint.Access{APIKey: GetAPIKey(r.Context()), Level: level} - } - // unauthenticated: one configured key makes the api key unambiguous - return endpoint.Access{APIKey: s.singleAPIKey} - }, - Logger: s.logger, - SingleKeyFallback: true, - }) -} - -func (s *AgentService) ServeHTTP(w http.ResponseWriter, r *http.Request) { - if conn, registration, ok := s.upgrader.Upgrade(w, r, nil); ok { - // control/signal conn: bound a single signalling frame (0 = operator - // disabled) before it is buffered; decompressed size is bounded in - // WSSignalConnection. HTTP endpoints are served over a worker's - // WebTransport session (see the WebTransport listener), not this - // WebSocket, which now carries control/job-dispatch traffic only. - if s.signalMessageSizeLimit > 0 { - conn.SetReadLimit(s.signalMessageSizeLimit) - } - sigConn := NewWSSignalConnection(conn, s.signalMessageSizeLimit) - defer sigConn.Close() - s.HandleConnection(r.Context(), sigConn, registration) - } -} - -func NewAgentHandler( - agentServer rpc.AgentInternalServer, - keyProvider auth.KeyProvider, - logger logger.Logger, - serverInfo *livekit.ServerInfo, - targetLoad float32, - roomTopic string, - publisherTopic string, - participantTopic string, -) *AgentHandler { - return &AgentHandler{ - agentServer: agentServer, - logger: logger.WithComponent("agents"), + registry *endpoint.Registry, +) (*AgentHandler, error) { + h := &AgentHandler{ + logger: logger.GetLogger().WithComponent("agents"), workers: make(map[string]*agent.Worker), jobToWorker: make(map[livekit.JobID]*agent.Worker), namespaceWorkers: make(map[workerKey][]*agent.Worker), - serverInfo: serverInfo, + serverInfo: &livekit.ServerInfo{ + Edition: livekit.ServerInfo_Standard, + Version: version.Version, + Protocol: types.CurrentProtocol, + AgentProtocol: agent.CurrentProtocol, + Region: conf.Region, + NodeId: string(currentNode.NodeID()), + }, keyProvider: keyProvider, - targetLoad: targetLoad, - roomTopic: roomTopic, - publisherTopic: publisherTopic, - participantTopic: participantTopic, - endpointRegistry: endpoint.NewRegistry(), + targetLoad: conf.Agents.TargetLoad, + roomTopic: agent.RoomAgentTopic, + publisherTopic: agent.PublisherAgentTopic, + participantTopic: agent.ParticipantAgentTopic, + endpointRegistry: registry, + endpointsConfig: conf.Agents.Endpoints, + } + if len(conf.Keys) == 1 { + for key := range conf.Keys { + h.singleAPIKey = key + } } -} -// endpointSettings validates a registration's manifest and negotiates the -// data-plane protocol version. -func (h *AgentHandler) endpointSettings(req *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error) { - if h.endpointsConfig.Disabled { - return nil, errors.New("agent HTTP endpoints are disabled on this server") - } - if _, err := endpoint.ParseManifest(req.GetEndpoints()); err != nil { + agentServer, err := rpc.NewAgentInternalServer(h, bus) + if err != nil { return nil, err } - if req.GetInstanceId() == "" { - return nil, errors.New("registrations with endpoints require an instance_id") + h.agentServer = agentServer + + return h, nil +} + +// endpointRegisterHandler negotiates a worker's endpoint settings, refusing any +// declaration when endpoints are turned off. +func (h *AgentHandler) endpointRegisterHandler(req *livekit.RegisterWorkerRequest, res *livekit.RegisterWorkerResponse, reg *agent.WorkerRegistration) error { + if len(req.GetEndpoints()) > 0 && h.endpointsConfig.Disabled { + return errors.New("agent HTTP endpoints are disabled on this server") } - if req.GetEndpointProtocol() == 0 { - return nil, errors.New("worker declared endpoints but no endpoint protocol; upgrade the agent SDK to one that speaks the endpoint data plane") - } - // the worker frames to the version returned here, so it must be the - // negotiated one and not the server's own constant - negotiated := min(req.GetEndpointProtocol(), wire.CurrentProtocol) - if negotiated < wire.MinProtocol { - return nil, fmt.Errorf("unsupported agent endpoint protocol %d (this server serves %d..%d)", - req.GetEndpointProtocol(), wire.MinProtocol, wire.CurrentProtocol) - } - return &livekit.AgentHttp_AgentEndpointSettings{Protocol: negotiated}, nil + return agent.EndpointRegisterHandler(req, res, reg) } // HandleConnection serves a worker's control connection with no data-plane @@ -324,7 +177,7 @@ func (h *AgentHandler) HandleConnection(ctx context.Context, conn agent.SignalCo // exchanges share it); nil for a WebSocket control connection, which serves no // endpoints. func (h *AgentHandler) handleConnection(ctx context.Context, conn agent.SignalConn, registration agent.WorkerRegistration, sess endpoint.Session) { - registration, ok := HandshakeAgentWorker(conn, h.serverInfo, registration, h.logger, h.endpointSettings) + registration, ok := HandshakeAgentWorker(conn, h.serverInfo, registration, h.logger, h.endpointRegisterHandler) if !ok { return } @@ -513,6 +366,7 @@ func (h *AgentHandler) JobRequest(ctx context.Context, job *livekit.Job) (*rpc.J "jobID", job.Id, "namespace", job.Namespace, "agentName", job.AgentName, + "deployment", job.Deployment, "jobType", job.Type.String(), ) if job.Room != nil { diff --git a/pkg/service/agentwebtransport.go b/pkg/service/agentwebtransport.go deleted file mode 100644 index 0d8091946..000000000 --- a/pkg/service/agentwebtransport.go +++ /dev/null @@ -1,230 +0,0 @@ -// 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 ( - "context" - "crypto/ecdsa" - "crypto/elliptic" - "crypto/rand" - "crypto/tls" - "crypto/x509" - "crypto/x509/pkix" - "errors" - "math/big" - "net" - "net/http" - "strconv" - "sync" - "time" - - "github.com/quic-go/quic-go/http3" - "github.com/quic-go/webtransport-go" - - "github.com/livekit/livekit-server/pkg/agent" - "github.com/livekit/livekit-server/pkg/agent/endpoint" - "github.com/livekit/livekit-server/pkg/agent/endpoint/wire" - "github.com/livekit/livekit-server/pkg/rtc" - "github.com/livekit/protocol/auth" - "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/logger" -) - -// StartWebTransport starts the worker WebTransport listener (control + data) on -// the configured UDP port, if endpoints are enabled and a port is set. It -// returns the stop function, nil when no listener was started. QUIC has no -// plaintext mode, so a TLS certificate is required: from -// tls_cert_file/tls_key_file, or a generated self-signed cert in dev mode. -func (s *AgentService) StartWebTransport() (func(), error) { - cfg := s.endpointsConfig - if cfg.Disabled || cfg.WebTransportPort == 0 { - return nil, nil - } - tlsConf, err := WebTransportTLS(cfg.TLSCertFile, cfg.TLSKeyFile, s.developmentMode) - if err != nil { - return nil, err - } - udp, err := net.ListenUDP("udp", &net.UDPAddr{Port: int(cfg.WebTransportPort)}) - if err != nil { - return nil, err - } - wt := NewAgentWebTransportServer(s, s.keyProvider, tlsConf) - go func() { - if err := wt.Serve(udp); err != nil { - logger.Infow("agent webtransport listener stopped", "error", err) - } - }() - logger.Infow("agent webtransport listener started", "port", cfg.WebTransportPort) - return func() { _ = wt.Close(); _ = udp.Close() }, nil -} - -// WebTransportTLS builds the listener's TLS config from cert files, or a -// generated self-signed cert in dev mode. -func WebTransportTLS(certFile, keyFile string, dev bool) (*tls.Config, error) { - if certFile != "" && keyFile != "" { - cert, err := tls.LoadX509KeyPair(certFile, keyFile) - if err != nil { - return nil, err - } - return &tls.Config{Certificates: []tls.Certificate{cert}, NextProtos: []string{http3.NextProtoH3}}, nil - } - if !dev { - return nil, errors.New("agent endpoints WebTransport requires agents.endpoints.tls_cert_file/tls_key_file") - } - key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) - if err != nil { - return nil, err - } - tmpl := &x509.Certificate{ - SerialNumber: big.NewInt(1), - Subject: pkix.Name{CommonName: "livekit-agent-webtransport-dev"}, - NotBefore: time.Now().Add(-time.Hour), - NotAfter: time.Now().Add(365 * 24 * time.Hour), - IPAddresses: []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback}, - DNSNames: []string{"localhost"}, - } - der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) - if err != nil { - return nil, err - } - logger.Warnw("agent webtransport using a generated self-signed cert (dev mode); workers must skip verification", nil) - return &tls.Config{ - Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}, - NextProtos: []string{http3.NextProtoH3}, - }, nil -} - -// NewWebTransportServer wraps an HTTP/3 WebTransport server around tlsConf (the -// h3 ALPN is set here if absent). The caller must assign wt.H3.Handler: routes -// that Upgrade need the *webtransport.Server itself. -func NewWebTransportServer(tlsConf *tls.Config) *webtransport.Server { - tlsConf = tlsConf.Clone() - if len(tlsConf.NextProtos) == 0 { - tlsConf.NextProtos = []string{http3.NextProtoH3} - } - return &webtransport.Server{H3: &http3.Server{TLSConfig: tlsConf}} -} - -// NewAgentWebTransportServer builds the WebTransport server that terminates a -// worker's unified session on /agent: the control stream carries the same -// WorkerMessage/ServerMessage exchange as the WebSocket control connection, and -// each HTTP exchange rides a node-opened QUIC stream on the same session. The -// handler runs behind the same api-key auth middleware as the rest of the -// agent surface, so the agent grant is enforced identically. -func NewAgentWebTransportServer(svc *AgentService, keyProvider auth.KeyProvider, tlsConf *tls.Config) *webtransport.Server { - wt := NewWebTransportServer(tlsConf) - authMW := NewAPIKeyAuthMiddleware(keyProvider) - mux := http.NewServeMux() - mux.HandleFunc("/agent", func(w http.ResponseWriter, r *http.Request) { - svc.ServeWebTransport(wt, w, r) - }) - wt.H3.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - authMW.ServeHTTP(w, r, mux.ServeHTTP) - }) - return wt -} - -// ServeWebTransport verifies the agent grant, upgrades the request to a -// WebTransport session, and serves it. It assumes an outer auth middleware has -// already placed the grant in the request context (as the WebSocket path does). -func (s *AgentService) ServeWebTransport(wt *webtransport.Server, w http.ResponseWriter, r *http.Request) { - claims := GetGrants(r.Context()) - if claims == nil || claims.Video == nil || !claims.Video.Agent { - HandleError(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied) - return - } - apiKey := GetAPIKey(r.Context()) - - registration := agent.MakeWorkerRegistration() - registration.ClientIP = GetClientIP(r) - if pv, err := strconv.Atoi(r.FormValue("protocol")); err == nil { - registration.Protocol = agent.WorkerProtocolVersion(pv) - } - - sess, err := wt.Upgrade(w, r) - if err != nil { - s.logger.Warnw("agent webtransport upgrade failed", err) - return - } - - // a detached context carries the worker's identity for the whole session; - // the control loop exits on the control stream's read error when the session - // dies, so it needs no cancellation from the request. - ctx := WithGrants(context.Background(), claims, apiKey) - go s.serveWebTransportSession(ctx, sess, registration) -} - -func (s *AgentService) serveWebTransportSession(ctx context.Context, sess *webtransport.Session, registration agent.WorkerRegistration) { - // the worker opens the control stream first (see the conformance worker and - // the SDK); every other stream the node opens is an HTTP exchange. Bound the - // wait so a session that upgrades but never opens a control stream can't hold - // a goroutine until the QUIC idle timeout. - acceptCtx, cancel := context.WithTimeout(sess.Context(), agent.RegisterTimeout) - defer cancel() - control, err := sess.AcceptStream(acceptCtx) - if err != nil { - _ = sess.CloseWithError(wire.SessionCloseOK, "no control stream") - return - } - sigConn := NewWTSignalConn(sess, control) - defer sigConn.Close() - maxStreams := int(s.endpointsConfig.MaxStreams) - if maxStreams <= 0 { - maxStreams = endpoint.DefaultMaxStreams - } - epSession := endpoint.NewWebTransportSession(sess, maxStreams) - s.handleConnection(ctx, sigConn, registration, epSession) -} - -// wtSignalConn adapts a WebTransport control stream to agent.SignalConn: the -// same length-delimited WorkerMessage/ServerMessage exchange the WebSocket -// control connection carries, framed on one QUIC bidirectional stream. -type wtSignalConn struct { - sess *webtransport.Session - control *webtransport.Stream - writeMu sync.Mutex -} - -// NewWTSignalConn adapts a WebTransport session's control stream to -// agent.SignalConn. -func NewWTSignalConn(sess *webtransport.Session, control *webtransport.Stream) agent.SignalConn { - return &wtSignalConn{sess: sess, control: control} -} - -func (c *wtSignalConn) WriteServerMessage(msg *livekit.ServerMessage) (int, error) { - c.writeMu.Lock() - defer c.writeMu.Unlock() - return 0, wire.WriteControlMessage(c.control, msg) -} - -func (c *wtSignalConn) ReadWorkerMessage() (*livekit.WorkerMessage, int, error) { - var msg livekit.WorkerMessage - if err := wire.ReadControlMessage(c.control, &msg); err != nil { - return nil, 0, err - } - return &msg, 0, nil -} - -func (c *wtSignalConn) SetReadDeadline(t time.Time) error { - return c.control.SetReadDeadline(t) -} - -func (c *wtSignalConn) Close() error { - return c.sess.CloseWithError(wire.SessionCloseOK, "") -} - -func (c *wtSignalConn) CloseWithReason(reason string) error { - return c.sess.CloseWithError(wire.SessionCloseOK, reason) -} diff --git a/pkg/service/agentws.go b/pkg/service/agentws.go new file mode 100644 index 000000000..c9f0c538a --- /dev/null +++ b/pkg/service/agentws.go @@ -0,0 +1,102 @@ +// Copyright 2024 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 ( + "net/http" + "strconv" + + "github.com/gorilla/websocket" + + "github.com/livekit/livekit-server/pkg/agent" + "github.com/livekit/livekit-server/pkg/config" + "github.com/livekit/livekit-server/pkg/rtc" +) + +type AgentSocketUpgrader struct { + websocket.Upgrader +} + +func (u AgentSocketUpgrader) Upgrade( + w http.ResponseWriter, + r *http.Request, + responseHeader http.Header, +) ( + conn *websocket.Conn, + registration agent.WorkerRegistration, + ok bool, +) { + if u.CheckOrigin == nil { + // allow connections from any origin, since script may be hosted anywhere + // security is enforced by access tokens + u.CheckOrigin = func(r *http.Request) bool { + return true + } + } + + if !websocket.IsWebSocketUpgrade(r) { + w.WriteHeader(404) + return + } + + claims := GetGrants(r.Context()) + if claims == nil || claims.Video == nil || !claims.Video.Agent { + HandleError(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied) + return + } + + registration = agent.MakeWorkerRegistration() + registration.ClientIP = GetClientIP(r) + + conn, err := u.Upgrader.Upgrade(w, r, responseHeader) + if err != nil { + HandleError(w, r, http.StatusInternalServerError, err) + return + } + + if pv, err := strconv.Atoi(r.FormValue("protocol")); err == nil { + registration.Protocol = agent.WorkerProtocolVersion(pv) + } + + return conn, registration, true +} + +// AgentWSService serves a worker's WebSocket control connection on /agent: +// control and job dispatch only, with no data-plane session for HTTP endpoints. +type AgentWSService struct { + *AgentHandler + + upgrader AgentSocketUpgrader + signalMessageSizeLimit int64 +} + +func NewAgentWSService(conf *config.Config, h *AgentHandler) *AgentWSService { + return &AgentWSService{ + AgentHandler: h, + signalMessageSizeLimit: conf.Limit.AgentSignalMessageSizeLimit, + } +} + +func (s *AgentWSService) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if conn, registration, ok := s.upgrader.Upgrade(w, r, nil); ok { + // bound a single signalling frame before it is buffered; 0 disables + if s.signalMessageSizeLimit > 0 { + conn.SetReadLimit(s.signalMessageSizeLimit) + } + sigConn := NewWSSignalConnection(conn, s.signalMessageSizeLimit) + defer sigConn.Close() + s.HandleConnection(r.Context(), sigConn, registration) + } +} diff --git a/pkg/service/agentwt.go b/pkg/service/agentwt.go new file mode 100644 index 000000000..d442eb4db --- /dev/null +++ b/pkg/service/agentwt.go @@ -0,0 +1,143 @@ +// 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 ( + "context" + "net/http" + "strconv" + "sync" + "time" + + "github.com/quic-go/webtransport-go" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + + "github.com/livekit/livekit-server/pkg/agent" + "github.com/livekit/livekit-server/pkg/agent/endpoint" + "github.com/livekit/livekit-server/pkg/agent/endpoint/wire" + "github.com/livekit/livekit-server/pkg/rtc" +) + +// AgentWTService serves a worker's WebTransport session on /agent: one control +// stream, plus a node-opened QUIC stream per HTTP exchange. +type AgentWTService struct { + *AgentHandler +} + +func NewAgentWTService(h *AgentHandler) *AgentWTService { + return &AgentWTService{AgentHandler: h} +} + +func (s *AgentWTService) ServeHTTP(w http.ResponseWriter, r *http.Request) { + sess, registration, ok := UpgradeAgentWebTransport(w, r, s.logger) + if !ok { + return + } + + sigConn, epSession, ok := AcceptAgentControlStream(sess, s.endpointsConfig.MaxStreams) + if !ok { + return + } + defer sigConn.Close() + + s.handleConnection(r.Context(), sigConn, registration, epSession) +} + +// UpgradeAgentWebTransport verifies the agent grant, builds the worker's +// registration from the request and upgrades to a WebTransport session. An outer +// auth middleware must already have placed the grant in the request context. +func UpgradeAgentWebTransport(w http.ResponseWriter, r *http.Request, l logger.Logger) ( + sess *webtransport.Session, + registration agent.WorkerRegistration, + ok bool, +) { + claims := GetGrants(r.Context()) + if claims == nil || claims.Video == nil || !claims.Video.Agent { + HandleError(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied) + return + } + + registration = agent.MakeWorkerRegistration() + registration.ClientIP = GetClientIP(r) + if pv, err := strconv.Atoi(r.FormValue("protocol")); err == nil { + registration.Protocol = agent.WorkerProtocolVersion(pv) + } + + sess, err := UpgradeWebTransport(w, r) + if err != nil { + l.Warnw("agent webtransport upgrade failed", err) + return nil, registration, false + } + + return sess, registration, true +} + +// AcceptAgentControlStream waits for the worker's control stream and wraps the +// session for the data plane. The wait is bounded so a session that upgrades but +// never opens a control stream can't hold a goroutine until the QUIC idle timeout. +func AcceptAgentControlStream(sess *webtransport.Session, maxStreams uint32) (agent.SignalConn, endpoint.Session, bool) { + acceptCtx, cancel := context.WithTimeout(sess.Context(), agent.RegisterTimeout) + defer cancel() + control, err := sess.AcceptStream(acceptCtx) + if err != nil { + _ = sess.CloseWithError(wire.SessionCloseOK, "no control stream") + return nil, nil, false + } + if maxStreams == 0 { + maxStreams = endpoint.DefaultMaxStreams + } + return NewWTSignalConn(sess, control), endpoint.NewWebTransportSession(sess, int(maxStreams)), true +} + +// wtSignalConn adapts a WebTransport control stream to agent.SignalConn: a +// length-delimited WorkerMessage/ServerMessage exchange on one QUIC bidirectional +// stream. +type wtSignalConn struct { + sess *webtransport.Session + control *webtransport.Stream + writeMu sync.Mutex +} + +func NewWTSignalConn(sess *webtransport.Session, control *webtransport.Stream) agent.SignalConn { + return &wtSignalConn{sess: sess, control: control} +} + +func (c *wtSignalConn) WriteServerMessage(msg *livekit.ServerMessage) (int, error) { + c.writeMu.Lock() + defer c.writeMu.Unlock() + return 0, wire.WriteControlMessage(c.control, msg) +} + +func (c *wtSignalConn) ReadWorkerMessage() (*livekit.WorkerMessage, int, error) { + var msg livekit.WorkerMessage + if err := wire.ReadControlMessage(c.control, &msg); err != nil { + return nil, 0, err + } + return &msg, 0, nil +} + +func (c *wtSignalConn) SetReadDeadline(t time.Time) error { + return c.control.SetReadDeadline(t) +} + +func (c *wtSignalConn) Close() error { + return c.sess.CloseWithError(wire.SessionCloseOK, "") +} + +func (c *wtSignalConn) CloseWithReason(reason string) error { + return c.sess.CloseWithError(wire.SessionCloseOK, reason) +} diff --git a/pkg/service/server.go b/pkg/service/server.go index a8749a16c..a648ca348 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -16,6 +16,7 @@ package service import ( "context" + "crypto/tls" "encoding/json" "errors" "fmt" @@ -29,6 +30,7 @@ import ( "github.com/pion/turn/v5" "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/quic-go/webtransport-go" "github.com/rs/cors" "github.com/twitchtv/twirp" "github.com/urfave/negroni/v3" @@ -46,23 +48,22 @@ import ( ) type LivekitServer struct { - config *config.Config - ioService *IOInfoService - rtcService *RTCService - whipService *WHIPService - agentService *AgentService - httpServer *http.Server - promServer *http.Server - debugServer *http.Server - router routing.Router - roomManager *RoomManager - signalServer *SignalServer - turnServer *turn.Server - currentNode routing.LocalNode - agentWTStop func() - running atomic.Bool - doneChan chan struct{} - closedChan chan struct{} + config *config.Config + ioService *IOInfoService + rtcService *RTCService + whipService *WHIPService + httpServer *http.Server + promServer *http.Server + debugServer *http.Server + webtransportServer *webtransport.Server + router routing.Router + roomManager *RoomManager + signalServer *SignalServer + turnServer *turn.Server + currentNode routing.LocalNode + running atomic.Bool + doneChan chan struct{} + closedChan chan struct{} } func NewLivekitServer(conf *config.Config, @@ -74,7 +75,9 @@ func NewLivekitServer(conf *config.Config, ioService *IOInfoService, rtcService *RTCService, whipService *WHIPService, - agentService *AgentService, + agentWSService *AgentWSService, + agentWTService *AgentWTService, + agentEndpointService *AgentEndpointService, keyProvider auth.KeyProvider, router routing.Router, roomManager *RoomManager, @@ -87,7 +90,6 @@ func NewLivekitServer(conf *config.Config, ioService: ioService, rtcService: rtcService, whipService: whipService, - agentService: agentService, router: router, roomManager: roomManager, signalServer: signalServer, @@ -128,12 +130,13 @@ func NewLivekitServer(conf *config.Config, xtwirp.RegisterServer(mux, sipServer) rtcService.SetupRoutes(mux) whipService.SetupRoutes(mux) - mux.Handle("/agent", agentService) + mux.Handle("/agent", agentWSService) mux.HandleFunc("/", s.defaultHandler) + // NewHTTPHandler branches on a nil handler, which a typed-nil would defeat var agentFront http.Handler if !conf.Agents.Endpoints.Disabled { - agentFront = agentService.EndpointFront() + agentFront = agentEndpointService } s.httpServer = &http.Server{ @@ -176,6 +179,19 @@ func NewLivekitServer(conf *config.Config, } } + if conf.WebTransport.Port > 0 { + var tlsConf *tls.Config + tlsConf, err = WebTransportTLS(conf.WebTransport.TLSCertFile, conf.WebTransport.TLSKeyFile, conf.Development) + if err != nil { + return + } + wtMux := http.NewServeMux() + wtMux.Handle("/agent", agentWTService) + + s.webtransportServer = NewWebTransportServer(tlsConf) + s.webtransportServer.H3.Handler = NewWebTransportHandler(keyProvider, s.webtransportServer, wtMux) + } + if err = router.RemoveDeadNodes(); err != nil { return } @@ -218,14 +234,6 @@ func (s *LivekitServer) Start() error { return err } - if s.agentService != nil { - stop, err := s.agentService.StartWebTransport() - if err != nil { - return err - } - s.agentWTStop = stop - } - addresses := s.config.BindAddresses if addresses == nil { addresses = []string{""} @@ -259,6 +267,15 @@ func (s *LivekitServer) Start() error { } } + stopWebTransport := func() {} + if s.webtransportServer != nil { + _, stop, err := ListenWebTransport(s.webtransportServer, s.config.BindAddresses, s.config.WebTransport.Port) + if err != nil { + return err + } + stopWebTransport = stop + } + values := []any{ "portHttp", s.config.Port, "nodeID", s.currentNode.NodeID(), @@ -284,6 +301,9 @@ func (s *LivekitServer) Start() error { if s.config.DebugHandler.Port != 0 { values = append(values, "portDebugHandler", s.config.DebugHandler.Port) } + if s.config.WebTransport.Port != 0 { + values = append(values, "portWebTransport", s.config.WebTransport.Port) + } if s.config.Region != "" { values = append(values, "region", s.config.Region) } @@ -334,6 +354,7 @@ func (s *LivekitServer) Start() error { if s.debugServer != nil { _ = s.debugServer.Shutdown(ctx) } + stopWebTransport() if s.turnServer != nil { _ = s.turnServer.Close() @@ -364,9 +385,6 @@ func (s *LivekitServer) Stop(force bool) { } s.router.Stop() - if s.agentWTStop != nil { - s.agentWTStop() - } close(s.doneChan) // wait for fully closed diff --git a/pkg/service/webtransport.go b/pkg/service/webtransport.go new file mode 100644 index 000000000..0a7431939 --- /dev/null +++ b/pkg/service/webtransport.go @@ -0,0 +1,177 @@ +// 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 ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "errors" + "math/big" + "net" + "net/http" + "strconv" + "time" + + "github.com/quic-go/quic-go/http3" + "github.com/quic-go/webtransport-go" + "github.com/urfave/negroni/v3" + + "github.com/livekit/protocol/auth" + "github.com/livekit/protocol/logger" +) + +// WebTransportTLS builds the listener's TLS config from cert files, or a +// generated self-signed cert in dev mode. +func WebTransportTLS(certFile, keyFile string, dev bool) (*tls.Config, error) { + if certFile != "" && keyFile != "" { + cert, err := tls.LoadX509KeyPair(certFile, keyFile) + if err != nil { + return nil, err + } + return &tls.Config{Certificates: []tls.Certificate{cert}, NextProtos: []string{http3.NextProtoH3}}, nil + } + if !dev { + return nil, errors.New("webtransport requires webtransport.tls_cert_file/tls_key_file") + } + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, err + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "livekit-webtransport-dev"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(365 * 24 * time.Hour), + IPAddresses: []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback}, + DNSNames: []string{"localhost"}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + return nil, err + } + logger.Warnw("webtransport using a generated self-signed cert (dev mode); clients must skip verification", nil) + return &tls.Config{ + Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}, + NextProtos: []string{http3.NextProtoH3}, + }, nil +} + +// NewWebTransportServer wraps an HTTP/3 WebTransport server around tlsConf (the +// h3 ALPN is set here if absent). The caller must assign wt.H3.Handler. +func NewWebTransportServer(tlsConf *tls.Config) *webtransport.Server { + tlsConf = tlsConf.Clone() + if len(tlsConf.NextProtos) == 0 { + tlsConf.NextProtos = []string{http3.NextProtoH3} + } + return &webtransport.Server{H3: &http3.Server{TLSConfig: tlsConf}} +} + +// NewWebTransportHandler builds the listener's handler: mux behind api-key auth, +// with wt in each request's context. +func NewWebTransportHandler(keyProvider auth.KeyProvider, wt *webtransport.Server, mux http.Handler) http.Handler { + middlewares := []negroni.Handler{negroni.NewRecovery()} + if keyProvider != nil { + middlewares = append(middlewares, NewAPIKeyAuthMiddleware(keyProvider)) + } + return WithWebTransportServer(wt, WithPathNormalization(configureMiddlewares(mux, middlewares...))) +} + +type webTransportServerKey struct{} + +// WithWebTransportServer puts wt in each request's context, where a route that +// upgrades reads it. Apply it outermost, ahead of the listener's middleware chain. +func WithWebTransportServer(wt *webtransport.Server, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), webTransportServerKey{}, wt))) + }) +} + +// GetWebTransportServer returns the server serving this request, or nil when the +// request did not arrive over a WebTransport listener. +func GetWebTransportServer(ctx context.Context) *webtransport.Server { + wt, _ := ctx.Value(webTransportServerKey{}).(*webtransport.Server) + return wt +} + +// UpgradeWebTransport upgrades a request on the server carried in its context. +func UpgradeWebTransport(w http.ResponseWriter, r *http.Request) (*webtransport.Session, error) { + wt := GetWebTransportServer(r.Context()) + if wt == nil { + return nil, errors.New("no webtransport server serving this request") + } + return wt.Upgrade(unwrapResponseWriter(w), r) +} + +// unwrapResponseWriter reaches the ResponseWriter net/http handed in. An upgrade +// type-asserts it to http3.Settingser and http3.HTTPStreamer without checking, so +// it must not be a middleware's wrapper. +func unwrapResponseWriter(w http.ResponseWriter) http.ResponseWriter { + for { + u, ok := w.(interface{ Unwrap() http.ResponseWriter }) + if !ok { + return w + } + w = u.Unwrap() + } +} + +// ListenWebTransport binds a UDP socket per address and serves wt on each, +// returning the bound addresses and a stop func. Empty addrs binds all interfaces. +func ListenWebTransport(wt *webtransport.Server, addrs []string, port uint32) ([]net.Addr, func(), error) { + if len(addrs) == 0 { + addrs = []string{""} + } + + conns := make([]*net.UDPConn, 0, len(addrs)) + bound := make([]net.Addr, 0, len(addrs)) + closeAll := func() { + for _, c := range conns { + _ = c.Close() + } + } + for _, addr := range addrs { + udpAddr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(addr, strconv.Itoa(int(port)))) + if err != nil { + closeAll() + return nil, nil, err + } + udp, err := net.ListenUDP("udp", udpAddr) + if err != nil { + closeAll() + return nil, nil, err + } + conns = append(conns, udp) + bound = append(bound, udp.LocalAddr()) + } + + for _, udp := range conns { + go func() { + if err := wt.Serve(udp); err != nil { + logger.Infow("webtransport listener stopped", "error", err) + } + }() + } + logger.Infow("webtransport listener started", "addresses", bound) + + return bound, func() { + _ = wt.Close() + closeAll() + }, nil +} diff --git a/pkg/service/webtransport_test.go b/pkg/service/webtransport_test.go new file mode 100644 index 000000000..2af6c6006 --- /dev/null +++ b/pkg/service/webtransport_test.go @@ -0,0 +1,47 @@ +// 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 ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/require" + "github.com/urfave/negroni/v3" +) + +// An upgrade type-asserts the ResponseWriter to http3.Settingser and +// http3.HTTPStreamer without checking, so it must reach the one net/http handed +// in. Every negroni chain wraps it. +func TestUnwrapResponseWriterReachesTheOriginal(t *testing.T) { + original := httptest.NewRecorder() + + var seen http.ResponseWriter + chain := negroni.New() + chain.UseHandler(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + seen = w + })) + chain.ServeHTTP(original, httptest.NewRequest(http.MethodGet, "/agent", nil)) + + _, wrapped := seen.(interface{ Unwrap() http.ResponseWriter }) + require.True(t, wrapped, "negroni no longer wraps; this test no longer covers anything") + require.Same(t, original, unwrapResponseWriter(seen)) +} + +func TestUnwrapResponseWriterPassesThroughUnwrapped(t *testing.T) { + original := httptest.NewRecorder() + require.Same(t, original, unwrapResponseWriter(original)) +} diff --git a/pkg/service/wire.go b/pkg/service/wire.go index 9d1842b4b..a6ba4bae0 100644 --- a/pkg/service/wire.go +++ b/pkg/service/wire.go @@ -38,6 +38,7 @@ import ( "github.com/livekit/psrpc/pkg/middleware/otelpsrpc" "github.com/livekit/livekit-server/pkg/agent" + "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/pkg/sfu" @@ -80,7 +81,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live NewRoomService, NewRTCService, NewWHIPService, - NewAgentService, + endpoint.NewRegistry, + NewAgentHandler, + NewAgentWSService, + NewAgentWTService, + NewAgentEndpointService, NewAgentDispatchService, getAgentConfig, agent.NewAgentClient, diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index f7c5a6267..46ca06c93 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -8,7 +8,15 @@ package service import ( "fmt" + "os" + + "github.com/pion/turn/v5" + "github.com/pkg/errors" + "github.com/redis/go-redis/v9" + "gopkg.in/yaml.v3" + "github.com/livekit/livekit-server/pkg/agent" + "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/pkg/sfu" @@ -22,11 +30,6 @@ import ( "github.com/livekit/protocol/webhook" "github.com/livekit/psrpc" "github.com/livekit/psrpc/pkg/middleware/otelpsrpc" - "github.com/pion/turn/v5" - "github.com/pkg/errors" - "github.com/redis/go-redis/v9" - "gopkg.in/yaml.v3" - "os" ) // Injectors from wire.go: @@ -125,10 +128,14 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live if err != nil { return nil, err } - agentService, err := NewAgentService(conf, currentNode, v, keyProvider) + registry := endpoint.NewRegistry() + agentHandler, err := NewAgentHandler(conf, currentNode, v, keyProvider, registry) if err != nil { return nil, err } + agentWSService := NewAgentWSService(conf, agentHandler) + agentWTService := NewAgentWTService(agentHandler) + agentEndpointService := NewAgentEndpointService(agentHandler, registry) agentConfig := getAgentConfig(conf) client, err := agent.NewAgentClient(v, agentConfig) if err != nil { @@ -151,7 +158,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live if err != nil { return nil, err } - livekitServer, err := NewLivekitServer(conf, roomService, agentDispatchService, egressService, ingressService, sipService, ioInfoService, rtcService, serviceWHIPService, agentService, keyProvider, router, roomManager, signalServer, server, currentNode) + livekitServer, err := NewLivekitServer(conf, roomService, agentDispatchService, egressService, ingressService, sipService, ioInfoService, rtcService, serviceWHIPService, agentWSService, agentWTService, agentEndpointService, keyProvider, router, roomManager, signalServer, server, currentNode) if err != nil { return nil, err }