fix(publisher): cap concurrent playback connections to prevent resource-exhaustion DoS

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"
                 }
             }
         },

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"
                 }
             }
         },

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:

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).

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

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
+	}
+}

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)")
+}

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 —

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.

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)