diff --git a/pkg/media/frame_server.go b/pkg/media/frame_server.go index 20b4965b6..6b2dad416 100644 --- a/pkg/media/frame_server.go +++ b/pkg/media/frame_server.go @@ -185,7 +185,7 @@ func ServeMKVIngestWorkerSocket(ctx context.Context, cfg IngestWorkerConfig, std } defer func() { ln.Close() - _ = os.Remove(cfg.SocketPath) + removeWorkerFiles(cfg.SocketPath) }() srv := newFrameServer(workerFrameBuffer) diff --git a/pkg/media/ingest_daemon.go b/pkg/media/ingest_daemon.go index b9a0977aa..4a60fba04 100644 --- a/pkg/media/ingest_daemon.go +++ b/pkg/media/ingest_daemon.go @@ -77,9 +77,53 @@ func SpawnIngestWorkerDetached(cfg IngestWorkerConfig, media *os.File) (*os.Proc _ = cmd.Process.Kill() return nil, fmt.Errorf("send worker config: %w", err) } + // Resume sidecar: a restarting main discovers sockets by UUID alone, which + // carries no identity. Record the streamer DID + PID next to the socket so the + // new main can re-arm ban enforcement (watchKeyRevocation) and, since it won't + // hold this worker's process handle, kill it by PID if needed. Best-effort: + // without it the worker still runs, just without restart-surviving ban + // enforcement. + if err := writeWorkerMeta(cfg.SocketPath, workerMeta{StreamerDID: cfg.StreamerDID, PID: cmd.Process.Pid}); err != nil { + log.Warn(context.Background(), "ingest worker: write resume metadata failed", "socket", cfg.SocketPath, "error", err) + } return cmd.Process, nil } +// workerMeta is the resume sidecar written next to a detached worker's socket: +// enough for a restarting main to enforce bans on a worker it didn't spawn. +type workerMeta struct { + StreamerDID string `json:"streamer_did"` + PID int `json:"pid"` +} + +func workerMetaPath(socketPath string) string { + return strings.TrimSuffix(socketPath, ".sock") + ".json" +} + +func writeWorkerMeta(socketPath string, meta workerMeta) error { + b, err := json.Marshal(meta) + if err != nil { + return err + } + return os.WriteFile(workerMetaPath(socketPath), b, 0o644) +} + +func readWorkerMeta(socketPath string) (workerMeta, error) { + var m workerMeta + b, err := os.ReadFile(workerMetaPath(socketPath)) + if err != nil { + return m, err + } + return m, json.Unmarshal(b, &m) +} + +// removeWorkerFiles unlinks a worker's socket and its resume sidecar together, +// so the sidecar never outlives the socket it describes. +func removeWorkerFiles(socketPath string) { + _ = os.Remove(socketPath) + _ = os.Remove(workerMetaPath(socketPath)) +} + // ConsumeWorkerSocket connects to a worker's frame socket and feeds its segments // to onSegment, reconnecting across transient disconnects. A worker that // outlived a main restart keeps buffering and replays on reconnect; ValidateMP4's @@ -96,7 +140,7 @@ func (mm *MediaManager) ConsumeWorkerSocket(ctx context.Context, socketPath, str // Never came up within the grace: a dead/never-started worker, or a // stale socket a SIGKILLed worker left behind. Unlink it so it can't // linger or re-spin a consumer on the next restart. - _ = os.Remove(socketPath) + removeWorkerFiles(socketPath) return fmt.Errorf("ingest worker never came up at %s: %w", socketPath, err) } select { // worker may still be coming up @@ -109,7 +153,7 @@ func (mm *MediaManager) ConsumeWorkerSocket(ctx context.Context, socketPath, str // Connected before, now the socket's gone: the worker exited. A clean // exit already unlinked its socket; a SIGKILL leaves it behind, so remove // the leftover (no-op if already gone). - _ = os.Remove(socketPath) + removeWorkerFiles(socketPath) return fmt.Errorf("ingest worker socket gone before End: %w", err) } connectedOnce = true @@ -189,7 +233,7 @@ func (mm *MediaManager) MKVIngestDetached(ctx context.Context, conn net.Conn, pr // bus/model), so main watches and kills it. proc.Kill (not ctx cancel) so the // reap below still fires — a ctx cancel means "main shutting down, leave it // running", which is the opposite of what a ban wants. - go mm.watchKeyRevocation(ctx, ms, func(reason string) { + go mm.watchKeyRevocation(ctx, ms.Streamer(), ms.DID(), func(reason string) { log.Warn(ctx, "detached ingest worker: ending stream", "reason", reason, "streamer", ms.Streamer()) _ = proc.Kill() }) @@ -306,7 +350,7 @@ func (mm *MediaManager) WHIPIngestDetached(ctx context.Context, offerSDP string, // Scoped to this consume's lifetime so it doesn't outlive the stream. wctx, wcancel := context.WithCancel(ctx) defer wcancel() - go mm.watchKeyRevocation(wctx, ms, func(reason string) { + go mm.watchKeyRevocation(wctx, ms.Streamer(), ms.DID(), func(reason string) { log.Warn(wctx, "whip ingest worker: ending stream", "reason", reason, "streamer", ms.Streamer()) _ = proc.Kill() }) @@ -341,9 +385,27 @@ func (mm *MediaManager) ResumeDetachedWorkers(ctx context.Context) { } for _, sock := range socks { sock := sock - log.Log(ctx, "resuming detached ingest worker", "socket", sock) + meta, merr := readWorkerMeta(sock) + streamer := "resumed" + if merr == nil && meta.StreamerDID != "" { + streamer = meta.StreamerDID + } + log.Log(ctx, "resuming detached ingest worker", "socket", sock, "streamer", streamer) go func() { - cerr := mm.ConsumeWorkerSocket(ctx, sock, "resumed", mm.validateSegment(ctx)) + wctx, wcancel := context.WithCancel(ctx) + defer wcancel() + // Re-arm ban enforcement for a worker we didn't spawn. We have no process + // handle, so kill by the PID in the sidecar (guarded against PID reuse). + // Without metadata we can still drain the worker, just not enforce bans. + if merr != nil || meta.StreamerDID == "" { + log.Warn(ctx, "resumed ingest worker: no resume metadata; cannot enforce bans on it", "socket", sock, "error", merr) + } else { + go mm.watchKeyRevocation(wctx, meta.StreamerDID, meta.StreamerDID, func(reason string) { + log.Warn(wctx, "resumed ingest worker: ending stream", "reason", reason, "streamer", meta.StreamerDID) + killWorkerPID(wctx, meta.PID, meta.StreamerDID) + }) + } + cerr := mm.ConsumeWorkerSocket(wctx, sock, streamer, mm.validateSegment(ctx)) if cerr != nil { log.Error(ctx, "resumed ingest worker ended", "socket", sock, "error", cerr) } diff --git a/pkg/media/ingest_isolation_linux.go b/pkg/media/ingest_isolation_linux.go index f318c738d..602e06480 100644 --- a/pkg/media/ingest_isolation_linux.go +++ b/pkg/media/ingest_isolation_linux.go @@ -3,8 +3,14 @@ package media import ( + "bytes" + "context" + "fmt" + "os" "os/exec" "syscall" + + "stream.place/streamplace/pkg/log" ) // IngestIsolationSupported reports whether per-stream ingest isolation (worker @@ -21,3 +27,35 @@ func IngestIsolationSupported() bool { return true } func setDetached(cmd *exec.Cmd) { cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true} } + +// killWorkerPID SIGKILLs a resumed detached worker by PID. A restarting main +// doesn't hold the process handle of a worker a previous main spawned, so it +// kills by the number recorded in the resume sidecar — but only after confirming +// via /proc//cmdline that the PID is still OUR `ingest-worker ` (the +// DID rides argv). That guards against PID reuse: if the worker already died and +// its number was recycled, the cmdline won't match and we leave the new process +// alone. +func killWorkerPID(ctx context.Context, pid int, did string) { + if pid <= 0 { + return + } + cmdline, err := os.ReadFile(fmt.Sprintf("/proc/%d/cmdline", pid)) + if err != nil { + return // process already gone + } + args := bytes.Split(cmdline, []byte{0}) // argv is NUL-separated + ours := false + for i := 0; i+1 < len(args); i++ { + if string(args[i]) == "ingest-worker" && string(args[i+1]) == did { + ours = true + break + } + } + if !ours { + log.Warn(ctx, "resumed worker PID is not our ingest-worker (reused?); not killing", "pid", pid, "did", did) + return + } + if err := syscall.Kill(pid, syscall.SIGKILL); err != nil { + log.Error(ctx, "failed to kill resumed ingest worker", "pid", pid, "error", err) + } +} diff --git a/pkg/media/ingest_isolation_notlinux.go b/pkg/media/ingest_isolation_notlinux.go index 58fa155ba..c5f39cd44 100644 --- a/pkg/media/ingest_isolation_notlinux.go +++ b/pkg/media/ingest_isolation_notlinux.go @@ -2,7 +2,10 @@ package media -import "os/exec" +import ( + "context" + "os/exec" +) // IngestIsolationSupported is false off Linux: the worker-subprocess transport // relies on Unix fd-passing (exec.Cmd.ExtraFiles, unsupported on Windows) and @@ -15,3 +18,7 @@ func IngestIsolationSupported() bool { return false } // this only needs to keep the build compiling (no Setsid field exists on // Windows' syscall.SysProcAttr). func setDetached(cmd *exec.Cmd) {} + +// killWorkerPID is a no-op off Linux: the detached/resume path is gated off +// there (ResumeDetachedWorkers never runs), so this only keeps the build green. +func killWorkerPID(ctx context.Context, pid int, did string) {} diff --git a/pkg/media/ingest_subprocess_test.go b/pkg/media/ingest_subprocess_test.go index 04d518d05..f9f46d9b3 100644 --- a/pkg/media/ingest_subprocess_test.go +++ b/pkg/media/ingest_subprocess_test.go @@ -24,6 +24,13 @@ import ( // config on fd 3, frames on fd 4, MKV on stdin; clean run ends with End, a fatal // error with an Error frame and a non-zero exit. func runIngestWorkerHelper() int { + // Test hook: a worker-shaped process that just sleeps (same argv layout as a + // real worker, so /proc//cmdline matches killWorkerPID's safety check). + // Lets the kill / resume-ban tests act on a real PID without a gst pipeline. + if len(os.Args) > 3 && os.Args[3] == "__test_sleep__" { + time.Sleep(10 * time.Minute) + return 0 + } cfgFile := os.NewFile(3, "ingest-config") if cfgFile == nil { return 2 diff --git a/pkg/media/ingest_supervisor.go b/pkg/media/ingest_supervisor.go index 9bf22b4d8..5f09a9849 100644 --- a/pkg/media/ingest_supervisor.go +++ b/pkg/media/ingest_supervisor.go @@ -139,7 +139,7 @@ func (mm *MediaManager) MKVIngestIsolated(ctx context.Context, input io.Reader, // behalf and kills it (cancel → CommandContext SIGKILLs the worker), matching // the in-process HandleKeyRevocation. Without this a banned user keeps // streaming under --isolated-ingest. - go mm.watchKeyRevocation(ctx, ms, func(reason string) { + go mm.watchKeyRevocation(ctx, ms.Streamer(), ms.DID(), func(reason string) { log.Warn(ctx, "ingest worker: ending stream", "reason", reason, "streamer", ms.Streamer()) cancel() }) diff --git a/pkg/media/key_revocation.go b/pkg/media/key_revocation.go index 53b421ed8..21dcf13f6 100644 --- a/pkg/media/key_revocation.go +++ b/pkg/media/key_revocation.go @@ -12,14 +12,17 @@ import ( // watchKeyRevocation blocks until the streamer's signing key is revoked or the // streamer is banned (or ctx ends), then calls onRevoked once with a reason and -// returns. It is the shared core behind both the in-process HandleKeyRevocation -// (which errors the gst pipeline) and the isolated-worker supervisors (which -// kill the worker subprocess). The latter matters because an isolated worker has -// no bus/model of its own, so it cannot notice a ban — main has to watch on its -// behalf and tear the worker down, or a banned user keeps streaming. -func (mm *MediaManager) watchKeyRevocation(ctx context.Context, ms MediaSigner, onRevoked func(reason string)) { - sub := mm.bus.Subscribe(ms.Streamer()) - defer mm.bus.Unsubscribe(ms.Streamer(), sub) +// returns. It takes the streamer's bus key + DID as plain strings (not a +// MediaSigner) so it can also be driven from just the resume metadata of a +// worker that outlived a main restart. It is the shared core behind both the +// in-process HandleKeyRevocation (which errors the gst pipeline) and the +// isolated-worker supervisors (which kill the worker subprocess). The latter +// matters because an isolated worker has no bus/model of its own, so it cannot +// notice a ban — main has to watch on its behalf, or a banned user keeps +// streaming. +func (mm *MediaManager) watchKeyRevocation(ctx context.Context, streamer, did string, onRevoked func(reason string)) { + sub := mm.bus.Subscribe(streamer) + defer mm.bus.Unsubscribe(streamer, sub) for { select { case <-ctx.Done(): @@ -27,7 +30,7 @@ func (mm *MediaManager) watchKeyRevocation(ctx context.Context, ms MediaSigner, case msg := <-sub: switch v := msg.(type) { case *model.SigningKey: - if v.RevokedAt != nil && v.DID == ms.DID() { + if v.RevokedAt != nil && v.DID == did { onRevoked(fmt.Sprintf("signing key revoked: %s", v.RKey)) return } @@ -44,7 +47,7 @@ func (mm *MediaManager) watchKeyRevocation(ctx context.Context, ms MediaSigner, // HandleKeyRevocation shuts down an in-process ingest pipeline when the // streamer's signing key is revoked or they get banned. func (mm *MediaManager) HandleKeyRevocation(ctx context.Context, ms MediaSigner, pipeline *gst.Pipeline) { - mm.watchKeyRevocation(ctx, ms, func(reason string) { + mm.watchKeyRevocation(ctx, ms.Streamer(), ms.DID(), func(reason string) { err := fmt.Errorf("ending stream: %s", reason) pipeline.Error(err.Error(), err) }) diff --git a/pkg/media/key_revocation_linux_test.go b/pkg/media/key_revocation_linux_test.go new file mode 100644 index 000000000..86b801c73 --- /dev/null +++ b/pkg/media/key_revocation_linux_test.go @@ -0,0 +1,120 @@ +//go:build linux + +package media + +import ( + "bytes" + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "syscall" + "testing" + "time" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/atproto" + "stream.place/streamplace/pkg/bus" + "stream.place/streamplace/pkg/config" +) + +// spawnFakeWorker re-execs the test binary as a worker-shaped process that just +// sleeps — same argv as a real worker (".../test ingest-worker ") so +// /proc//cmdline matches killWorkerPID's safety check, without needing a gst +// pipeline. +func spawnFakeWorker(t *testing.T, did string) *exec.Cmd { + t.Helper() + exe, err := os.Executable() + require.NoError(t, err) + cmd := exec.Command(exe, "ingest-worker", did, "__test_sleep__") + require.NoError(t, cmd.Start()) + t.Cleanup(func() { + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + }) + // Wait for exec() to complete so /proc//cmdline reflects the worker argv — + // the property killWorkerPID's safety check reads. + require.Eventually(t, func() bool { + b, rerr := os.ReadFile(fmt.Sprintf("/proc/%d/cmdline", cmd.Process.Pid)) + return rerr == nil && bytes.Contains(b, []byte("ingest-worker\x00"+did)) + }, 5*time.Second, 20*time.Millisecond, "fake worker did not exec with the expected cmdline") + return cmd +} + +// TestKillWorkerPIDRefusesMismatch is the PID-reuse guard: a PID whose +// /proc cmdline is NOT our ingest-worker for this DID must be left alone. +func TestKillWorkerPIDRefusesMismatch(t *testing.T) { + cmd := spawnFakeWorker(t, "did:plc:realstreamer") + pid := cmd.Process.Pid + + killWorkerPID(context.Background(), pid, "did:plc:someone-else") + + time.Sleep(300 * time.Millisecond) + // Signal 0 probes liveness; the process wasn't killed, so it's running (not a + // zombie), and this returns nil. + require.NoError(t, syscall.Kill(pid, 0), "killWorkerPID must not kill a PID whose cmdline DID doesn't match") +} + +// TestKillWorkerPIDKillsMatch: the matching worker is killed. +func TestKillWorkerPIDKillsMatch(t *testing.T) { + did := "did:plc:realstreamer" + cmd := spawnFakeWorker(t, did) + + died := make(chan struct{}) + go func() { _, _ = cmd.Process.Wait(); close(died) }() + + killWorkerPID(context.Background(), cmd.Process.Pid, did) + + select { + case <-died: + case <-time.After(5 * time.Second): + t.Fatal("killWorkerPID did not kill the matching worker") + } +} + +// TestResumeDetachedWorkerBanContained closes the resumed-worker gap end to end: +// a detached worker that outlived a main restart (here a worker-shaped sleeper +// plus its socket + resume sidecar) gets torn down when its streamer is banned, +// even though the new main never spawned it and holds no process handle. +func TestResumeDetachedWorkerBanContained(t *testing.T) { + did := "did:plc:resumed-streamer" + mm := &MediaManager{bus: bus.NewBus(), cli: &config.CLI{DataDir: t.TempDir()}} + + dir, err := mm.ingestWorkerSocketDir() + require.NoError(t, err) + + // Stand in for a worker left behind by a previous main: a worker-shaped + // sleeper, plus a discovered socket file and the resume sidecar pointing at it. + worker := spawnFakeWorker(t, did) + sockPath := filepath.Join(dir, uuid.NewString()+".sock") + require.NoError(t, os.WriteFile(sockPath, nil, 0o644)) // discovered; not a live listener + require.NoError(t, writeWorkerMeta(sockPath, workerMeta{StreamerDID: did, PID: worker.Process.Pid})) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Resume reads the sidecar and re-arms ban enforcement while it (fails to) + // drain the dead socket within the connect grace. + mm.ResumeDetachedWorkers(ctx) + + died := make(chan struct{}) + go func() { _, _ = worker.Process.Wait(); close(died) }() + + banned := &comatproto.LabelDefs_Label{Val: atproto.LabelDMCAViolation, Uri: did} + tick := time.NewTicker(100 * time.Millisecond) + defer tick.Stop() + deadline := time.After(15 * time.Second) + for { + select { + case <-died: + return // the resumed worker was killed by the ban — gap closed + case <-deadline: + t.Fatal("resumed worker was not killed after its streamer was banned") + case <-tick.C: + mm.bus.Publish(did, banned) // re-publish until the watcher's subscription is live + } + } +} diff --git a/pkg/media/key_revocation_test.go b/pkg/media/key_revocation_test.go index 1646722e5..fcfc05333 100644 --- a/pkg/media/key_revocation_test.go +++ b/pkg/media/key_revocation_test.go @@ -24,7 +24,7 @@ func TestWatchKeyRevocationBan(t *testing.T) { defer cancel() revoked := make(chan string, 1) - go mm.watchKeyRevocation(ctx, ms, func(reason string) { revoked <- reason }) + go mm.watchKeyRevocation(ctx, ms.Streamer(), ms.DID(), func(reason string) { revoked <- reason }) banned := &comatproto.LabelDefs_Label{Val: atproto.LabelDMCAViolation, Uri: "did:plc:test-streamer"} tick := time.NewTicker(50 * time.Millisecond) diff --git a/pkg/media/whip_worker.go b/pkg/media/whip_worker.go index 09bb221e2..f8270b8ac 100644 --- a/pkg/media/whip_worker.go +++ b/pkg/media/whip_worker.go @@ -39,7 +39,7 @@ func ServeWHIPIngestWorkerSocket(ctx context.Context, cfg IngestWorkerConfig) er } defer func() { ln.Close() - _ = os.Remove(cfg.SocketPath) + removeWorkerFiles(cfg.SocketPath) }() srv := newFrameServer(workerFrameBuffer)