diff --git a/pkg/media/validate.go b/pkg/media/validate.go index 83097e835..ae3af324c 100644 --- a/pkg/media/validate.go +++ b/pkg/media/validate.go @@ -16,6 +16,7 @@ import ( "go.opentelemetry.io/otel/trace" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/aqtime" + "stream.place/streamplace/pkg/atproto" c2patypes "stream.place/streamplace/pkg/c2patypes" "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/crypto/signers" @@ -154,6 +155,21 @@ func (mm *MediaManager) validateSource(ctx context.Context, buf []byte, local bo return nil, fmt.Errorf("got valid segment, but user %s is not allowed: %w", repoDID, err) } + // Defense in depth: a banned streamer's ingest worker is torn down + // (watchKeyRevocation), but if that ever misses — a raced ban, a failed kill, + // a worker that somehow lived — this is the chokepoint every ingest path + // converges on, so refusing here keeps banned content from being distributed + // regardless. + _, labelSpan := tracer.Start(ctx, "ValidateMP4.streamerIsBanned") + banned, err := mm.streamerIsBanned(repoDID) + labelSpan.End() + if err != nil { + return nil, fmt.Errorf("check labels for %s: %w", repoDID, err) + } + if banned { + return nil, fmt.Errorf("got valid segment, but user %s is banned", repoDID) + } + // Apply content filtering after metadata is parsed if mm.cli.ContentFilters != nil { if err := mm.applyContentFilters(ctx, meta); err != nil { @@ -171,6 +187,22 @@ func (mm *MediaManager) validateSource(ctx context.Context, buf []byte, local bo }, nil } +// streamerIsBanned reports whether repoDID currently carries an active ban +// label. It's the defense-in-depth gate validateSource applies so a banned +// streamer's segments are refused even if their ingest worker wasn't torn down. +// Returns false when there's no model (the minimal worker/test managers); +// enforcement runs in main, which has the model + label feed. +func (mm *MediaManager) streamerIsBanned(repoDID string) (bool, error) { + if mm.model == nil { + return false, nil + } + labels, err := mm.model.GetActiveLabels(repoDID) + if err != nil { + return false, err + } + return atproto.IsBanned(labels...), nil +} + // distributeSegment archives the segment, folds it into the streamer's live-HLS // window, and notifies subscribers. seg is the bytes to store/distribute — the // completed dual-codec segment when completion ran, else the validated source diff --git a/pkg/media/validate_bare_test.go b/pkg/media/validate_bare_test.go index f254f0760..7852a8490 100644 --- a/pkg/media/validate_bare_test.go +++ b/pkg/media/validate_bare_test.go @@ -9,13 +9,17 @@ import ( "os" "sort" "testing" + "time" + comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/decred/dcrd/dcrec/secp256k1" "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/atproto" "stream.place/streamplace/pkg/crypto/aqpub" "stream.place/streamplace/pkg/crypto/signers" "stream.place/streamplace/pkg/livehls" + "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/muxl" ) @@ -153,3 +157,41 @@ func TestFeedLiveWindow(t *testing.T) { } require.Contains(t, w.MasterPlaylist(func(tid string) string { return tid + ".m3u8" }), "#EXTM3U") } + +// seedBanLabel writes an active ban label for did into the model. +func seedBanLabel(t *testing.T, mod model.Model, did string) { + t.Helper() + lex := &comatproto.LabelDefs_Label{ + Cts: time.Now().UTC().Format(time.RFC3339), + Src: "did:plc:test-labeler", + Uri: did, + Val: atproto.LabelDMCAViolation, + } + var buf bytes.Buffer + require.NoError(t, lex.MarshalCBOR(&buf)) + require.NoError(t, mod.CreateLabel(&model.Label{ + Src: lex.Src, + Uri: did, + Val: atproto.LabelDMCAViolation, + Record: buf.Bytes(), + })) +} + +// TestStreamerIsBanned exercises the defense-in-depth gate validateSource applies +// at the ingest chokepoint: a streamer with an active ban label is rejected +// regardless of whether their ingest worker was torn down. A clean streamer +// passes; the ONLY change between the two checks is the ban label. +func TestStreamerIsBanned(t *testing.T) { + mm, _ := getStaticTestMediaManager(t) + did := "did:plc:bannedstreamer" + + banned, err := mm.streamerIsBanned(did) + require.NoError(t, err) + require.False(t, banned, "a clean streamer is not banned") + + seedBanLabel(t, mm.model, did) + + banned, err = mm.streamerIsBanned(did) + require.NoError(t, err) + require.True(t, banned, "an active ban label is detected at the validate chokepoint") +}