diff --git a/api/docs/docs.go b/api/docs/docs.go index 324acdbd..3d566084 100644 --- a/api/docs/docs.go +++ b/api/docs/docs.go @@ -2199,6 +2199,13 @@ const docTemplate = `{ }, "hls": { "$ref": "#/definitions/config.PublisherHLSConfig" + }, + "max_playback_conn_per_stream": { + "description": "MaxPlaybackConnPerStream / MaxPlaybackConnTotal cap concurrent RTSP /\nRTMP / SRT / HTTP-MPEGTS playback connections (per stream and globally)\nto bound the heavyweight per-connection state a flood can allocate (A-1).\nUnset (0) applies a protective default; a negative value means unlimited.\nHLS/DASH viewers go over stateless HTTP and are not counted here.", + "type": "integer" + }, + "max_playback_conn_total": { + "type": "integer" } } }, diff --git a/api/docs/swagger.json b/api/docs/swagger.json index 26e027e9..827bd88a 100644 --- a/api/docs/swagger.json +++ b/api/docs/swagger.json @@ -2192,6 +2192,13 @@ }, "hls": { "$ref": "#/definitions/config.PublisherHLSConfig" + }, + "max_playback_conn_per_stream": { + "description": "MaxPlaybackConnPerStream / MaxPlaybackConnTotal cap concurrent RTSP /\nRTMP / SRT / HTTP-MPEGTS playback connections (per stream and globally)\nto bound the heavyweight per-connection state a flood can allocate (A-1).\nUnset (0) applies a protective default; a negative value means unlimited.\nHLS/DASH viewers go over stateless HTTP and are not counted here.", + "type": "integer" + }, + "max_playback_conn_total": { + "type": "integer" } } }, diff --git a/api/docs/swagger.yaml b/api/docs/swagger.yaml index ffd463a2..b1d5daf0 100644 --- a/api/docs/swagger.yaml +++ b/api/docs/swagger.yaml @@ -425,6 +425,16 @@ definitions: $ref: '#/definitions/config.PublisherDASHConfig' hls: $ref: '#/definitions/config.PublisherHLSConfig' + max_playback_conn_per_stream: + description: |- + MaxPlaybackConnPerStream / MaxPlaybackConnTotal cap concurrent RTSP / + RTMP / SRT / HTTP-MPEGTS playback connections (per stream and globally) + to bound the heavyweight per-connection state a flood can allocate (A-1). + Unset (0) applies a protective default; a negative value means unlimited. + HLS/DASH viewers go over stateless HTTP and are not counted here. + type: integer + max_playback_conn_total: + type: integer type: object config.PublisherDASHConfig: properties: diff --git a/config/config.go b/config/config.go index 4f8324b5..5ebffa59 100644 --- a/config/config.go +++ b/config/config.go @@ -97,6 +97,14 @@ type BufferConfig struct { type PublisherConfig struct { HLS PublisherHLSConfig `mapstructure:"hls" json:"hls" yaml:"hls"` DASH PublisherDASHConfig `mapstructure:"dash" json:"dash" yaml:"dash"` + + // MaxPlaybackConnPerStream / MaxPlaybackConnTotal cap concurrent RTSP / + // RTMP / SRT / HTTP-MPEGTS playback connections (per stream and globally) + // to bound the heavyweight per-connection state a flood can allocate (A-1). + // Unset (0) applies a protective default; a negative value means unlimited. + // HLS/DASH viewers go over stateless HTTP and are not counted here. + MaxPlaybackConnPerStream int `mapstructure:"max_playback_conn_per_stream" json:"max_playback_conn_per_stream" yaml:"max_playback_conn_per_stream"` + MaxPlaybackConnTotal int `mapstructure:"max_playback_conn_total" json:"max_playback_conn_total" yaml:"max_playback_conn_total"` } // PublisherHLSConfig is filesystem + live packaging for Apple HLS (m3u8 + segments). diff --git a/docs/audit/security-quality-audit.md b/docs/audit/security-quality-audit.md index f2d18ad6..b8bac19a 100644 --- a/docs/audit/security-quality-audit.md +++ b/docs/audit/security-quality-audit.md @@ -368,6 +368,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper **Fix:** Add per-stream and global concurrent-playback caps checked *before* allocating any pipeline: atomic counters on the publisher Service, increment-and-test at the top of `srtHandleConnect` (return REJECT), `HandleRTMPPlay` (error), RTSP `OnSetup`/`OnPlay` (503), `HandleMPEGTS` (503); decrement on close. For RTMP specifically, consider one shared demux per stream fanned out instead of a full 16 MiB pipeline per connection. Combine with the auth from S-1. +> ✅ **FIXED** in `fix/playback-conn-cap` — new `connLimiter` (`conn_limiter.go`) on the publisher `Service` caps concurrent playback connections per-stream AND globally; `acquire` is called BEFORE any per-connection state is allocated and `release` on close. Wiring: `HandleMPEGTS` (acquire→503, `defer release`), `HandleRTMPPlay` (acquire→error, `defer release`), `srtHandleSubscribe` (acquire→close+return, `defer release` — moved to the subscribe path so the slot is leak-proof via defer), and RTSP `OnPlay` (acquire→503; release in `OnSessionClose` via the `rtspSessions` entry, with a repeated-PLAY guard so PAUSE→PLAY can't double-count and a nil-`ps` cap-only entry handled by the touch loop / `pollBytes`). Caps are configurable (`PublisherConfig.MaxPlaybackConnPerStream` / `MaxPlaybackConnTotal`): unset → protective default (256 / 4096), negative → unlimited. HLS/DASH viewers go over stateless HTTP and aren't counted. The acquire/release balance across all 4 handlers was adversarially audited (incl. against gortsplib's serialized-per-session / guaranteed-`OnSessionClose` lifecycle) — no leak or double-release. Tests: `TestConnLimiter_*` (per-stream/global cap, isolation, unlimited, no-underflow, concurrent `-race` balance), `TestResolvePlaybackCap`, `TestHandleMPEGTS_PlaybackCapRejectsAndReleases` (503 over cap + slot frees after close). (Auth, the other half, remains S-1.) + --- #### A-2 (HIGH) — Unbounded DVR timeshift window loads the entire archive into memory per profile diff --git a/internal/publisher/conn_limiter.go b/internal/publisher/conn_limiter.go new file mode 100644 index 00000000..62307447 --- /dev/null +++ b/internal/publisher/conn_limiter.go @@ -0,0 +1,95 @@ +package publisher + +// conn_limiter.go — concurrent playback connection caps (A-1). +// +// The long-lived play/push protocols (RTSP, RTMP, SRT, HTTP-MPEGTS) allocate +// heavyweight per-connection state — a demux pipeline, a multi-MB write / ts +// buffer, 2-3 goroutines and an FD — with no natural bound. None of the +// handlers authenticate (the `token` param is attribution-only), so an +// unauthenticated client that knows one active stream code can open many +// concurrent connections and exhaust goroutines / FDs / memory, OOM-killing the +// shared process and taking every stream on the host down. connLimiter rejects +// a connection BEFORE that state is allocated, capping both per-stream and +// global concurrency. + +import ( + "sync" + + "github.com/ntt0601zcoder/open-streamer/internal/domain" +) + +// Protective defaults applied when the operator leaves the cap unset (0). These +// are generous for the play/push protocols (HLS/DASH viewers go over stateless +// HTTP and are not counted here) while still bounding a flood. Tune via +// PublisherConfig; a negative configured value means "explicitly unlimited". +const ( + defaultMaxPlaybackConnPerStream = 256 + defaultMaxPlaybackConnTotal = 4096 +) + +// connLimiter bounds concurrent playback connections per stream and globally. +// A max of 0 means unlimited for that dimension. Safe for concurrent use. +type connLimiter struct { + mu sync.Mutex + perStream map[domain.StreamCode]int + total int + maxPerStream int + maxTotal int +} + +func newConnLimiter(maxPerStream, maxTotal int) *connLimiter { + return &connLimiter{ + perStream: make(map[domain.StreamCode]int), + maxPerStream: maxPerStream, + maxTotal: maxTotal, + } +} + +// acquire reserves one connection slot for code. It returns false WITHOUT +// reserving when either the per-stream or the global cap is already at its +// limit; the caller must then reject the connection. On true the caller MUST +// pair it with exactly one release(code) when the connection closes — wire it +// with defer at the acquire site so no exit path leaks a slot (a leaked slot +// permanently inflates the count and eventually rejects every client). +func (l *connLimiter) acquire(code domain.StreamCode) bool { + l.mu.Lock() + defer l.mu.Unlock() + if l.maxTotal > 0 && l.total >= l.maxTotal { + return false + } + if l.maxPerStream > 0 && l.perStream[code] >= l.maxPerStream { + return false + } + l.total++ + l.perStream[code]++ + return true +} + +// release returns one slot reserved by a prior successful acquire(code). Never +// call it for a connection whose acquire returned false. +func (l *connLimiter) release(code domain.StreamCode) { + l.mu.Lock() + defer l.mu.Unlock() + if l.perStream[code] <= 1 { + delete(l.perStream, code) + } else { + l.perStream[code]-- + } + if l.total > 0 { + l.total-- + } +} + +// resolvePlaybackCap maps a configured cap to its effective value: a positive +// value is used as-is, a negative value is "explicitly unlimited" (0), and an +// unset value (0) falls back to the protective default. +func resolvePlaybackCap(configured, def int) int { + switch { + case configured > 0: + return configured + case configured < 0: + return 0 + default: + return def + } +} diff --git a/internal/publisher/conn_limiter_test.go b/internal/publisher/conn_limiter_test.go new file mode 100644 index 00000000..45003237 --- /dev/null +++ b/internal/publisher/conn_limiter_test.go @@ -0,0 +1,86 @@ +package publisher + +import ( + "sync" + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/ntt0601zcoder/open-streamer/internal/domain" +) + +func TestConnLimiter_PerStreamCap(t *testing.T) { + t.Parallel() + l := newConnLimiter(2, 0) // per-stream 2, total unlimited + c := domain.StreamCode("s") + assert.True(t, l.acquire(c)) + assert.True(t, l.acquire(c)) + assert.False(t, l.acquire(c), "3rd connection exceeds the per-stream cap") + l.release(c) + assert.True(t, l.acquire(c), "a slot frees after release") +} + +func TestConnLimiter_GlobalCap(t *testing.T) { + t.Parallel() + l := newConnLimiter(0, 2) // per-stream unlimited, total 2 + assert.True(t, l.acquire("a")) + assert.True(t, l.acquire("b")) + assert.False(t, l.acquire("c"), "3rd connection exceeds the global cap") + l.release("a") + assert.True(t, l.acquire("c"), "a slot frees globally after release") +} + +func TestConnLimiter_PerStreamIsolated(t *testing.T) { + t.Parallel() + l := newConnLimiter(1, 0) + assert.True(t, l.acquire("a")) + assert.False(t, l.acquire("a"), "stream a is at its cap") + assert.True(t, l.acquire("b"), "the per-stream cap is per code, not shared") +} + +func TestConnLimiter_Unlimited(t *testing.T) { + t.Parallel() + l := newConnLimiter(0, 0) // both unlimited + for i := 0; i < 1000; i++ { + assert.True(t, l.acquire("s")) + } +} + +func TestConnLimiter_SpuriousReleaseDoesNotUnderflow(t *testing.T) { + t.Parallel() + l := newConnLimiter(1, 1) + l.release("s") // release with nothing acquired must not drive the count negative + assert.True(t, l.acquire("s"), "still exactly one slot after a spurious release") + assert.False(t, l.acquire("s")) +} + +// TestConnLimiter_Concurrent stresses acquire/release under -race and asserts +// the net count balances back to zero (no leak / no underflow). +func TestConnLimiter_Concurrent(t *testing.T) { + t.Parallel() + l := newConnLimiter(0, 0) + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 100; j++ { + if l.acquire("s") { + l.release("s") + } + } + }() + } + wg.Wait() + l.mu.Lock() + defer l.mu.Unlock() + assert.Equal(t, 0, l.total, "total must balance to zero") + assert.Empty(t, l.perStream, "per-stream map must be empty after all releases") +} + +func TestResolvePlaybackCap(t *testing.T) { + t.Parallel() + assert.Equal(t, 50, resolvePlaybackCap(50, 256), "positive configured value used as-is") + assert.Equal(t, 256, resolvePlaybackCap(0, 256), "unset (0) falls back to the default") + assert.Equal(t, 0, resolvePlaybackCap(-1, 256), "negative means explicitly unlimited (0)") +} diff --git a/internal/publisher/serve_mpegts.go b/internal/publisher/serve_mpegts.go index 49123bc6..438d03b4 100644 --- a/internal/publisher/serve_mpegts.go +++ b/internal/publisher/serve_mpegts.go @@ -80,6 +80,15 @@ func (s *Service) HandleMPEGTS() http.HandlerFunc { return } + // Cap concurrent playback connections before allocating any per-client + // state (A-1). Released on every return path via defer. + if !s.limiter.acquire(code) { + slog.Warn("publisher: mpegts connection rejected — playback cap reached", "stream_code", code, "remote", r.RemoteAddr) + http.Error(w, "too many connections", http.StatusServiceUnavailable) + return + } + defer s.limiter.release(code) + sub, err := s.buf.Subscribe(bufID) if err != nil { // Buffer was deleted between mpegtsTargetFor and Subscribe — diff --git a/internal/publisher/serve_mpegts_test.go b/internal/publisher/serve_mpegts_test.go index f483bea6..d180bdbb 100644 --- a/internal/publisher/serve_mpegts_test.go +++ b/internal/publisher/serve_mpegts_test.go @@ -204,6 +204,65 @@ func TestHandleMPEGTS_ClientDisconnectReleasesSubscriber(t *testing.T) { } } +// TestHandleMPEGTS_PlaybackCapRejectsAndReleases verifies A-1 end-to-end: with +// a per-stream cap of 1, a second concurrent connection is rejected with 503, +// and once the first connection closes the freed slot lets a new one through +// (the deferred release in the handler ran). +func TestHandleMPEGTS_PlaybackCapRejectsAndReleases(t *testing.T) { + t.Parallel() + + streamCode := domain.StreamCode("ch_cap") + buf := buffer.NewServiceForTesting(64) + buf.Create(streamCode) + bus := events.New(4, 64) + pub := publisher.NewServiceForTesting(config.PublisherConfig{ + HLS: config.PublisherHLSConfig{Dir: t.TempDir()}, + DASH: config.PublisherDASHConfig{Dir: t.TempDir()}, + MaxPlaybackConnPerStream: 1, + }, buf, bus) + require.NoError(t, pub.Start(context.Background(), &domain.Stream{ + Code: streamCode, + Protocols: &domain.OutputProtocols{MPEGTS: true}, + })) + t.Cleanup(func() { pub.Stop(streamCode) }) + + r := chi.NewRouter() + r.Get("/{code}/mpegts", pub.HandleMPEGTS()) + srv := httptest.NewServer(r) + defer srv.Close() + url := srv.URL + "/ch_cap/mpegts" + + // conn1 occupies the only slot (acquire happens before the 200 header). + ctx1, cancel1 := context.WithCancel(context.Background()) + req1, _ := http.NewRequestWithContext(ctx1, http.MethodGet, url, nil) + resp1, err := http.DefaultClient.Do(req1) + require.NoError(t, err) + require.Equal(t, http.StatusOK, resp1.StatusCode) + + // conn2 is over the per-stream cap → 503, no slot held. + resp2 := mustGet(t, url) + assert.Equal(t, http.StatusServiceUnavailable, resp2.StatusCode) + _ = resp2.Body.Close() + + // Release conn1; the deferred release frees the slot. + cancel1() + _, _ = io.Copy(io.Discard, resp1.Body) + _ = resp1.Body.Close() + + require.Eventually(t, func() bool { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() // release the probe's own slot before the next tick + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + return false + } + ok := resp.StatusCode == http.StatusOK + _ = resp.Body.Close() + return ok + }, 2*time.Second, 50*time.Millisecond, "slot must free after conn1 closes") +} + // makeTSPayload returns one 188-byte TS-shaped packet whose 5th byte is the // given marker. Used so the test can assert on packet contents flowing end- // to-end without bringing in a full TS builder. diff --git a/internal/publisher/serve_rtmp.go b/internal/publisher/serve_rtmp.go index 51f480cd..78603304 100644 --- a/internal/publisher/serve_rtmp.go +++ b/internal/publisher/serve_rtmp.go @@ -54,6 +54,14 @@ func (s *Service) HandleRTMPPlay( return fmt.Errorf("stream %q not active", key) } + // Cap concurrent playback connections before allocating the per-session + // demux pipeline + tsBuffer (A-1). Released on every return path via defer. + if !s.limiter.acquire(code) { + slog.Warn("publisher: RTMP play rejected — playback cap reached", "stream_code", code, "remote", info.RemoteAddr) + return fmt.Errorf("stream %q: playback connection cap reached", key) + } + defer s.limiter.release(code) + sub, err := s.buf.Subscribe(bufID) if err != nil { return fmt.Errorf("subscribe %q: %w", key, err) diff --git a/internal/publisher/serve_rtsp.go b/internal/publisher/serve_rtsp.go index 572dabf4..e9d6e7c5 100644 --- a/internal/publisher/serve_rtsp.go +++ b/internal/publisher/serve_rtsp.go @@ -226,6 +226,24 @@ func (h *rtspHandler) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo if code == "" { return &base.Response{StatusCode: base.StatusBadRequest}, nil } + + // A repeated PLAY on the same session (e.g. after PAUSE) is already counted + // and tracked — don't re-acquire the cap or overwrite the entry. + h.svc.rtspSessionsMu.Lock() + _, already := h.svc.rtspSessions[ctx.Session] + h.svc.rtspSessionsMu.Unlock() + if already { + return &base.Response{StatusCode: base.StatusOK}, nil + } + + // Cap concurrent playback connections before tracking the session (A-1). + // The rtspSessions entry (created below on success) is the release marker + // OnSessionClose uses, so it is stored even when the tracker is absent. + if !h.svc.limiter.acquire(code) { + slog.Warn("publisher: RTSP play rejected — playback cap reached", "stream_code", code) + return &base.Response{StatusCode: base.StatusServiceUnavailable}, nil + } + ua := "" if ctx.Request != nil { if v, ok := ctx.Request.Header["User-Agent"]; ok && len(v) > 0 { @@ -233,11 +251,9 @@ func (h *rtspHandler) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo } } sess := openRTSPSession(context.Background(), h.svc.tracker, code, ctx.Conn.NetConn().RemoteAddr(), ua) - if sess != nil { - h.svc.rtspSessionsMu.Lock() - h.svc.rtspSessions[ctx.Session] = &rtspClient{ps: sess} - h.svc.rtspSessionsMu.Unlock() - } + h.svc.rtspSessionsMu.Lock() + h.svc.rtspSessions[ctx.Session] = &rtspClient{ps: sess, code: code} + h.svc.rtspSessionsMu.Unlock() return &base.Response{StatusCode: base.StatusOK}, nil } @@ -260,8 +276,11 @@ func (h *rtspHandler) OnSessionClose(ctx *gortsplib.ServerHandlerOnSessionCloseC if !ok { return } - rc.pollBytes(ctx.Session) - rc.ps.close() + h.svc.limiter.release(rc.code) // pair the OnPlay acquire (A-1) + if rc.ps != nil { + rc.pollBytes(ctx.Session) + rc.ps.close() + } } // lookupStream matches a request path to a registered ServerStream. @@ -420,8 +439,9 @@ func runRTSPPipeline( // directly. type rtspClient struct { ps *playSession - gortspSess rtspStatsSource // captured for nil-safe pollBytes when the close ctx isn't handy - lastOutbound atomic.Uint64 // cumulative bytes seen on the previous poll + code domain.StreamCode // stream code, for releasing the playback cap on close (ps may be nil when no tracker) + gortspSess rtspStatsSource // captured for nil-safe pollBytes when the close ctx isn't handy + lastOutbound atomic.Uint64 // cumulative bytes seen on the previous poll } // rtspStatsSource is the subset of *gortsplib.ServerSession pollBytes @@ -448,6 +468,9 @@ func (c *rtspClient) pollBytes(ss rtspStatsSource) { if ss == nil { return } + if c.ps == nil { + return // cap-only entry (no sessions tracker) — nothing to credit + } stats := ss.Stats() if stats == nil { return @@ -475,7 +498,7 @@ func (s *Service) touchRTSPSessions(streamCode domain.StreamCode) { s.rtspSessionsMu.Lock() victims := make([]victim, 0, len(s.rtspSessions)) for ss, rc := range s.rtspSessions { - if rc != nil && rc.ps.streamCode == streamCode { + if rc != nil && rc.ps != nil && rc.ps.streamCode == streamCode { rc.gortspSess = ss victims = append(victims, victim{client: rc, ss: ss}) } diff --git a/internal/publisher/serve_srt.go b/internal/publisher/serve_srt.go index 659a7089..efa1ed69 100644 --- a/internal/publisher/serve_srt.go +++ b/internal/publisher/serve_srt.go @@ -117,6 +117,14 @@ func (s *Service) srtHandleSubscribe(ctx context.Context, conn srt.Conn) { return } + // Cap concurrent playback connections before allocating the subscriber + + // muxer state (A-1). The conn is closed by the deferred conn.Close above. + if !s.limiter.acquire(streamCode) { + slog.Warn("publisher: SRT play rejected — playback cap reached", "stream_code", streamCode, "remote", conn.RemoteAddr()) + return + } + defer s.limiter.release(streamCode) + sub, err := s.buf.Subscribe(bufID) if err != nil { slog.Warn("publisher: SRT subscribe failed", "stream_code", streamCode, "err", err) diff --git a/internal/publisher/service.go b/internal/publisher/service.go index 495990c6..364988e2 100644 --- a/internal/publisher/service.go +++ b/internal/publisher/service.go @@ -96,6 +96,10 @@ type Service struct { tracker sessions.Tracker m *metrics.Metrics + // limiter caps concurrent playback connections (per stream + global) for + // the long-lived play/push protocols — see conn_limiter.go (A-1). + limiter *connLimiter + mu sync.Mutex streams map[domain.StreamCode]*streamState @@ -148,11 +152,15 @@ func New(i do.Injector) (*Service, error) { } svc := &Service{ - cfg: pub, - buf: buf, - bus: bus, - tracker: tracker, - m: m, + cfg: pub, + buf: buf, + bus: bus, + tracker: tracker, + m: m, + limiter: newConnLimiter( + resolvePlaybackCap(pub.MaxPlaybackConnPerStream, defaultMaxPlaybackConnPerStream), + resolvePlaybackCap(pub.MaxPlaybackConnTotal, defaultMaxPlaybackConnTotal), + ), streams: make(map[domain.StreamCode]*streamState), mediaBuffer: make(map[domain.StreamCode]domain.StreamCode), rtspMounts: make(map[string]*gortsplib.ServerStream), @@ -172,9 +180,13 @@ func New(i do.Injector) (*Service, error) { // NewServiceForTesting creates a Service without DI, for use in integration tests. func NewServiceForTesting(cfg config.PublisherConfig, buf *buffer.Service, bus events.Bus) *Service { svc := &Service{ - cfg: cfg, - buf: buf, - bus: bus, + cfg: cfg, + buf: buf, + bus: bus, + limiter: newConnLimiter( + resolvePlaybackCap(cfg.MaxPlaybackConnPerStream, defaultMaxPlaybackConnPerStream), + resolvePlaybackCap(cfg.MaxPlaybackConnTotal, defaultMaxPlaybackConnTotal), + ), streams: make(map[domain.StreamCode]*streamState), mediaBuffer: make(map[domain.StreamCode]domain.StreamCode), rtspMounts: make(map[string]*gortsplib.ServerStream),