ingress: add opt-in support for udp:// URL pull ingress (#4810)

* ingress: add opt-in support for udp:// URL pull ingress

URL pull ingress previously accepted only http, https and srt source
URLs. Add udp:// as an accepted scheme, gated behind a new
`ingress.enable_udp_url_pull` config option that defaults to false, so
unauthenticated UDP sources are not reachable unless the operator opts
in.

Also add a counterfeiter fake for IngressLauncher and a test covering
the scheme validation matrix.

The ingress handler binds a local socket on the caller supplied address
and port instead of connecting out like the http and srt sources do.
Spell out what that means for operators in both the config field doc
comment and config-sample.yaml: caller controlled local port binding,
unauthenticated and spoofable input, and multicast relaying of traffic
on the handler's local network.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

config-sample.yaml

@@ -337,6 +337,15 @@ keys:
 #   rtmp_base_url: "rtmp://my.domain.com/live"
 #   # Prefix used to generate WHIP URLs for WHIP ingress.
 #   whip_base_url: "http://my.domain.com/whip"
+#   # Allow URL pull ingress from udp:// source URLs. Disabled by default.
+#   # Only enable this if you trust both the callers allowed to create ingresses and the network the
+#   # ingress handlers run on. Unlike an http or srt source url, a udp source url doesn't make the
+#   # handler connect out to the url host: the handler binds a local socket on the address and port
+#   # taken from the url, and joins the multicast group if one is given. This lets the caller choose
+#   # which local port the handler binds, feed the session unauthenticated and easily spoofed traffic,
+#   # and make the handler join arbitrary multicast groups and republish whatever it receives into a
+#   # room, using the ingress as a relay for streams on the handler's local network.
+#   enable_udp_url_pull: false
 
 # Region of the current node. Required if using regionaware node selector
 # region: us-west-2

pkg/config/config.go

@@ -439,6 +439,19 @@ func (l LimitConfig) CanAddDataBlob(dataBlobs []*livekit.DataBlob, toAdd *liveki
 type IngressConfig struct {
 	RTMPBaseURL string `yaml:"rtmp_base_url,omitempty"`
 	WHIPBaseURL string `yaml:"whip_base_url,omitempty"`
+	// Allow URL pull ingresses with a udp:// source URL. Disabled by default, and should only be
+	// enabled if both the callers allowed to create ingresses and the network the ingress handlers
+	// run on are trusted. Unlike an http or srt source url, a udp source url doesn't make the ingress
+	// handler connect out to the url host: the handler binds a local socket on the address and port
+	// taken from the url, and joins the multicast group if one is given. This lets the caller:
+	//   - Choose which local port the handler binds, potentially colliding with other services on
+	//     the host.
+	//   - Feed the session unauthenticated traffic. UDP is connectionless, so any host able to reach
+	//     that port can inject media, or spoof the sender address to disrupt a legitimate feed.
+	//   - Make the handler join arbitrary multicast groups and republish whatever it receives into a
+	//     room, using the ingress as a relay for streams on the handler's local network the caller has
+	//     no direct access to.
+	EnableUDPURLPull bool `yaml:"enable_udp_url_pull,omitempty"`
 }
 
 type SIPConfig struct{}

pkg/service/ingress.go

@@ -19,8 +19,6 @@ import (
 	"fmt"
 	"net/url"
 
-	"github.com/livekit/livekit-server/pkg/config"
-	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/protocol/ingress"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
@@ -28,8 +26,12 @@ import (
 	"github.com/livekit/protocol/utils"
 	"github.com/livekit/protocol/utils/guid"
 	"github.com/livekit/psrpc"
+
+	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/telemetry"
 )
 
+//counterfeiter:generate . IngressLauncher
 type IngressLauncher interface {
 	LaunchPullIngress(ctx context.Context, info *livekit.IngressInfo) (*livekit.IngressInfo, error)
 }
@@ -133,7 +135,13 @@ func (s *IngressService) CreateIngressWithUrl(ctx context.Context, urlStr string
 		if err != nil {
 			return nil, psrpc.NewError(psrpc.InvalidArgument, err)
 		}
-		if urlObj.Scheme != "http" && urlObj.Scheme != "https" && urlObj.Scheme != "srt" {
+		switch urlObj.Scheme {
+		case "http", "https", "srt":
+		case "udp":
+			if !s.conf.EnableUDPURLPull {
+				return nil, ingress.ErrInvalidIngress("udp url pull is not enabled")
+			}
+		default:
 			return nil, ingress.ErrInvalidIngress(fmt.Sprintf("invalid url scheme %s", urlObj.Scheme))
 		}
 		// Marshall the URL again for sanitization

pkg/service/ingress_test.go

@@ -0,0 +1,96 @@
+// 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_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/protocol/auth"
+	"github.com/livekit/protocol/livekit"
+
+	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/service"
+	"github.com/livekit/livekit-server/pkg/service/servicefakes"
+)
+
+func TestCreateURLPullIngressScheme(t *testing.T) {
+	newService := func(enableUDP bool) (*service.IngressService, *servicefakes.FakeIngressLauncher) {
+		launcher := &servicefakes.FakeIngressLauncher{}
+		launcher.LaunchPullIngressCalls(func(_ context.Context, info *livekit.IngressInfo) (*livekit.IngressInfo, error) {
+			return info, nil
+		})
+
+		svc := service.NewIngressServiceWithIngressLauncher(
+			&config.IngressConfig{EnableUDPURLPull: enableUDP},
+			"nodeID",
+			nil,
+			nil,
+			&servicefakes.FakeIngressStore{},
+			nil,
+			nil,
+			launcher,
+		)
+		return svc, launcher
+	}
+
+	adminCtx := func() context.Context {
+		return service.WithGrants(context.Background(), &auth.ClaimGrants{Video: &auth.VideoGrant{IngressAdmin: true}}, "")
+	}
+
+	createReq := func(url string) *livekit.CreateIngressRequest {
+		return &livekit.CreateIngressRequest{
+			InputType:           livekit.IngressInput_URL_INPUT,
+			Url:                 url,
+			RoomName:            "testroom",
+			ParticipantIdentity: "ingress",
+		}
+	}
+
+	t.Run("udp rejected when disabled", func(t *testing.T) {
+		svc, launcher := newService(false)
+
+		_, err := svc.CreateIngress(adminCtx(), createReq("udp://1.2.3.4:1234"))
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "udp url pull is not enabled")
+		require.Zero(t, launcher.LaunchPullIngressCallCount())
+	})
+
+	t.Run("udp accepted when enabled", func(t *testing.T) {
+		svc, launcher := newService(true)
+
+		info, err := svc.CreateIngress(adminCtx(), createReq("udp://1.2.3.4:1234"))
+		require.NoError(t, err)
+		require.Equal(t, "udp://1.2.3.4:1234", info.Url)
+		require.Equal(t, 1, launcher.LaunchPullIngressCallCount())
+	})
+
+	t.Run("other schemes unaffected by the udp option", func(t *testing.T) {
+		for _, url := range []string{"http://example.com/live", "https://example.com/live", "srt://1.2.3.4:1234"} {
+			svc, _ := newService(false)
+
+			info, err := svc.CreateIngress(adminCtx(), createReq(url))
+			require.NoError(t, err, url)
+			require.Equal(t, url, info.Url)
+		}
+
+		svc, _ := newService(true)
+		_, err := svc.CreateIngress(adminCtx(), createReq("rtsp://1.2.3.4/live"))
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "invalid url scheme rtsp")
+	})
+}

pkg/service/servicefakes/fake_ingress_launcher.go

@@ -0,0 +1,118 @@
+// Code generated by counterfeiter. DO NOT EDIT.
+package servicefakes
+
+import (
+	"context"
+	"sync"
+
+	"github.com/livekit/livekit-server/pkg/service"
+	"github.com/livekit/protocol/livekit"
+)
+
+type FakeIngressLauncher struct {
+	LaunchPullIngressStub        func(context.Context, *livekit.IngressInfo) (*livekit.IngressInfo, error)
+	launchPullIngressMutex       sync.RWMutex
+	launchPullIngressArgsForCall []struct {
+		arg1 context.Context
+		arg2 *livekit.IngressInfo
+	}
+	launchPullIngressReturns struct {
+		result1 *livekit.IngressInfo
+		result2 error
+	}
+	launchPullIngressReturnsOnCall map[int]struct {
+		result1 *livekit.IngressInfo
+		result2 error
+	}
+	invocations      map[string][][]interface{}
+	invocationsMutex sync.RWMutex
+}
+
+func (fake *FakeIngressLauncher) LaunchPullIngress(arg1 context.Context, arg2 *livekit.IngressInfo) (*livekit.IngressInfo, error) {
+	fake.launchPullIngressMutex.Lock()
+	ret, specificReturn := fake.launchPullIngressReturnsOnCall[len(fake.launchPullIngressArgsForCall)]
+	fake.launchPullIngressArgsForCall = append(fake.launchPullIngressArgsForCall, struct {
+		arg1 context.Context
+		arg2 *livekit.IngressInfo
+	}{arg1, arg2})
+	stub := fake.LaunchPullIngressStub
+	fakeReturns := fake.launchPullIngressReturns
+	fake.recordInvocation("LaunchPullIngress", []interface{}{arg1, arg2})
+	fake.launchPullIngressMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2)
+	}
+	if specificReturn {
+		return ret.result1, ret.result2
+	}
+	return fakeReturns.result1, fakeReturns.result2
+}
+
+func (fake *FakeIngressLauncher) LaunchPullIngressCallCount() int {
+	fake.launchPullIngressMutex.RLock()
+	defer fake.launchPullIngressMutex.RUnlock()
+	return len(fake.launchPullIngressArgsForCall)
+}
+
+func (fake *FakeIngressLauncher) LaunchPullIngressCalls(stub func(context.Context, *livekit.IngressInfo) (*livekit.IngressInfo, error)) {
+	fake.launchPullIngressMutex.Lock()
+	defer fake.launchPullIngressMutex.Unlock()
+	fake.LaunchPullIngressStub = stub
+}
+
+func (fake *FakeIngressLauncher) LaunchPullIngressArgsForCall(i int) (context.Context, *livekit.IngressInfo) {
+	fake.launchPullIngressMutex.RLock()
+	defer fake.launchPullIngressMutex.RUnlock()
+	argsForCall := fake.launchPullIngressArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *FakeIngressLauncher) LaunchPullIngressReturns(result1 *livekit.IngressInfo, result2 error) {
+	fake.launchPullIngressMutex.Lock()
+	defer fake.launchPullIngressMutex.Unlock()
+	fake.LaunchPullIngressStub = nil
+	fake.launchPullIngressReturns = struct {
+		result1 *livekit.IngressInfo
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *FakeIngressLauncher) LaunchPullIngressReturnsOnCall(i int, result1 *livekit.IngressInfo, result2 error) {
+	fake.launchPullIngressMutex.Lock()
+	defer fake.launchPullIngressMutex.Unlock()
+	fake.LaunchPullIngressStub = nil
+	if fake.launchPullIngressReturnsOnCall == nil {
+		fake.launchPullIngressReturnsOnCall = make(map[int]struct {
+			result1 *livekit.IngressInfo
+			result2 error
+		})
+	}
+	fake.launchPullIngressReturnsOnCall[i] = struct {
+		result1 *livekit.IngressInfo
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *FakeIngressLauncher) Invocations() map[string][][]interface{} {
+	fake.invocationsMutex.RLock()
+	defer fake.invocationsMutex.RUnlock()
+	copiedInvocations := map[string][][]interface{}{}
+	for key, value := range fake.invocations {
+		copiedInvocations[key] = value
+	}
+	return copiedInvocations
+}
+
+func (fake *FakeIngressLauncher) recordInvocation(key string, args []interface{}) {
+	fake.invocationsMutex.Lock()
+	defer fake.invocationsMutex.Unlock()
+	if fake.invocations == nil {
+		fake.invocations = map[string][][]interface{}{}
+	}
+	if fake.invocations[key] == nil {
+		fake.invocations[key] = [][]interface{}{}
+	}
+	fake.invocations[key] = append(fake.invocations[key], args)
+}
+
+var _ service.IngressLauncher = new(FakeIngressLauncher)