diff --git a/config-sample.yaml b/config-sample.yaml index 5c1c50a42..f07299972 100644 --- a/config-sample.yaml +++ b/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 diff --git a/pkg/config/config.go b/pkg/config/config.go index 2ef6a3efe..b46756adb 100644 --- a/pkg/config/config.go +++ b/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{} diff --git a/pkg/service/ingress.go b/pkg/service/ingress.go index 6441756fb..528309c68 100644 --- a/pkg/service/ingress.go +++ b/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 diff --git a/pkg/service/ingress_test.go b/pkg/service/ingress_test.go new file mode 100644 index 000000000..86d29d2e2 --- /dev/null +++ b/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") + }) +} diff --git a/pkg/service/servicefakes/fake_ingress_launcher.go b/pkg/service/servicefakes/fake_ingress_launcher.go new file mode 100644 index 000000000..06ed69e6b --- /dev/null +++ b/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)