From e35efddac774247953d8388e2e9fec9dfa76d7f6 Mon Sep 17 00:00:00 2001 From: CoffeeKnyte <67730400+CoffeeKnyte@users.noreply.github.com> Date: Wed, 29 Jul 2026 15:17:36 +0000 Subject: [PATCH] fix(audiobooks): enforce the stream kill switch on ABS byte-serving routes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Audiobookshelf-compat surface sits outside the monitoring and kill design entirely — neither architecture matrix mentioned it. Three routes pour full media and none consulted the kill switch: - /(abs/)api/items/{id}/file/{ino} and /download — bearer auth, no revocation check, no in-flight cut. - /(abs/)public/session/{sid}/track/{idx} — mounted outside bearerAuth (the session id is the capability); it held a transport marker but was unkillable. - /feed/{slug}/file/{ino} — no auth at all, the slug is the capability; invisible and unkillable, and closing a feed only blocked the next request rather than cutting a pour already in flight. Each surface now passes its real credential-issue time, because a user revocation is a cutoff, not a ban: it matches only streams whose credential predates it. ABS bearer tokens are stateless JWTs that OnUserSessionsRevoked does not delete, so passing request-entry time (as the jellycompat login path safely does) would have meant a user kill never refused a later ABS request. - bearerAuth carries the JWT's iat into ctxAuth; the authenticated file route uses it. - The public track uses the persisted playback session's StartedAt and passes the native session id so session-level kills land too, plus the shared metered writer for byte accounting. - The feed file uses the feed's CreatedAt for an owner cutoff — a feed opened before the revocation dies, one opened after re-authenticating serves — and arms the in-flight cut. The authenticated file route stays download-class: like the native and jellycompat download routes it is exempt from the live-stream cap and from streammonitor, and is covered by no download quota. It is now killable and explicitly documented rather than quietly invisible. Bringing all three download-class routes under one quota and one monitor record is tracked as a follow-up rather than adding a fourth per-route model here. Part of the stream monitoring & kill-switch epic. --- cmd/silo/main.go | 1 + internal/audiobooks/abs/file_handler.go | 30 ++- .../abs/file_handler_public_track_test.go | 2 +- internal/audiobooks/abs/handler.go | 10 + internal/audiobooks/abs/revocation_test.go | 204 ++++++++++++++++++ internal/audiobooks/abs/rss_feeds_handler.go | 10 + internal/audiobooks/service.go | 3 + 7 files changed, 258 insertions(+), 2 deletions(-) create mode 100644 internal/audiobooks/abs/revocation_test.go diff --git a/cmd/silo/main.go b/cmd/silo/main.go index 8784a9aa..60515b8d 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -2285,6 +2285,7 @@ func main() { Detail: absDetailSvc, SessionMgr: sessionMgr, SessionSyncer: deps.SessionSyncer, + Revocation: streamRevocation, } absH := audiobooksService.BuildABSHandler(absHDeps) deps.ABSHandler = absH diff --git a/internal/audiobooks/abs/file_handler.go b/internal/audiobooks/abs/file_handler.go index 4c1ec280..cc8f27b6 100644 --- a/internal/audiobooks/abs/file_handler.go +++ b/internal/audiobooks/abs/file_handler.go @@ -52,12 +52,25 @@ func trackInoFor(contentID string, fileIdx int) string { // - Serve the bytes directly with Range-request support via playback.ServeDirectPlay. // - Set Content-Disposition: attachment on /download paths to encourage // browser save-to-disk / mobile offline-save behaviour. +// +// This is a download-class route: like native and Jellyfin-compatible +// downloads it is exempt from the live-stream cap and streammonitor, and it is +// not yet covered by a download quota. It is still revocable before and during +// the file pour. func (h *Handler) handleFileStream(w http.ResponseWriter, r *http.Request) { a, ok := absAuthFrom(r) if !ok || a.UserID == "" { http.Error(w, "unauthorized", http.StatusUnauthorized) return } + userID, err := strconv.Atoi(a.UserID) + if err != nil || userID <= 0 { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + if h.deps.Revocation != nil && h.deps.Revocation.Refuse(w, "", userID, a.IssuedAt) { + return + } contentID := chi.URLParam(r, "libraryItemId") inoStr := chi.URLParam(r, "ino") @@ -115,6 +128,8 @@ func (h *Handler) handleFileStream(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", ct) } + stop := h.deps.Revocation.WatchAndCut(w, "", userID, a.IssuedAt) + defer stop() if err := playback.ServeDirectPlay(w, r, mediaFile.FilePath); err != nil { // ServeDirectPlay has already written an error response; just log. return @@ -172,6 +187,14 @@ func (h *Handler) handlePublicTrack(w http.ResponseWriter, r *http.Request) { http.Error(w, "session expired", http.StatusGone) return } + userID, err := strconv.Atoi(sess.UserID) + if err != nil || userID <= 0 { + http.Error(w, "session owner invalid", http.StatusForbidden) + return + } + if h.deps.Revocation != nil && h.deps.Revocation.Refuse(w, sid, userID, sess.StartedAt) { + return + } if h.beginNativePlaybackTransport(sid) { defer h.endNativePlaybackTransport(sid) } @@ -196,7 +219,12 @@ func (h *Handler) handlePublicTrack(w http.ResponseWriter, r *http.Request) { if ct := audioContentType(ext); ct != "" { w.Header().Set("Content-Type", ct) } - _ = playback.ServeDirectPlay(w, r, mediaFile.FilePath) + recorder, _ := h.deps.NativeSessions.(playback.ServedBytesRecorder) + metered := playback.NewSessionMeteredWriter(w, recorder, sid) + defer func() { _ = metered.Close() }() + stop := h.deps.Revocation.WatchAndCut(metered, sid, userID, sess.StartedAt) + defer stop() + _ = playback.ServeDirectPlay(metered, r, mediaFile.FilePath) } func publicTrackSessionExpired(sess ABSPlaybackSession, now time.Time) bool { diff --git a/internal/audiobooks/abs/file_handler_public_track_test.go b/internal/audiobooks/abs/file_handler_public_track_test.go index 37931ddb..f8c051cd 100644 --- a/internal/audiobooks/abs/file_handler_public_track_test.go +++ b/internal/audiobooks/abs/file_handler_public_track_test.go @@ -88,7 +88,7 @@ func newPublicTrackHandler(t *testing.T, sid, contentID string, closed bool) (*H t.Helper() audioPath := makeTempAudio(t) sessStore := &fakePlaybackSessionStore{} - sess := ABSPlaybackSession{ID: sid, UserID: "u1", ContentID: contentID} + sess := ABSPlaybackSession{ID: sid, UserID: "1", ContentID: contentID, StartedAt: time.Now()} if closed { now := time.Now() sess.ClosedAt = &now diff --git a/internal/audiobooks/abs/handler.go b/internal/audiobooks/abs/handler.go index b615a8f7..d1b41d0b 100644 --- a/internal/audiobooks/abs/handler.go +++ b/internal/audiobooks/abs/handler.go @@ -21,6 +21,7 @@ import ( "github.com/Silo-Server/silo-server/internal/catalog" "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/streamrevoke" ) // --------------------------------------------------------------------------- @@ -277,6 +278,9 @@ type Dependencies struct { // see Audiobookshelf-compatible clients. May be nil; ABS playback still // functions, but admin live-session visibility is unavailable. NativeSessions PlaybackSessionManager + // Revocation makes ABS byte-serving routes obey the shared stream kill + // switch. Nil preserves compatibility for isolated tests. + Revocation *streamrevoke.Store // NativeSessionSyncer flushes native session-manager state into the shared // admin live-session table after ABS play/sync/close events. NativeSessionSyncer PlaybackSessionSyncer @@ -569,6 +573,7 @@ type ctxAuth struct { ProfileID string JTI string Token string // raw bearer token + IssuedAt time.Time } // absAuthFrom extracts ABS auth from the request context. Returns (zero, false) @@ -682,11 +687,16 @@ func (h *Handler) bearerAuth(next http.Handler) http.Handler { return } _ = h.deps.TokenStore.TouchToken(r.Context(), claims.JTI) + var issuedAt time.Time + if claims.IssuedAt != nil { + issuedAt = claims.IssuedAt.Time + } ctx := context.WithValue(r.Context(), ctxKey{}, ctxAuth{ UserID: claims.UserID, ProfileID: claims.ProfileID, JTI: claims.JTI, Token: raw, + IssuedAt: issuedAt, }) next.ServeHTTP(w, r.WithContext(ctx)) }) diff --git a/internal/audiobooks/abs/revocation_test.go b/internal/audiobooks/abs/revocation_test.go new file mode 100644 index 00000000..c2d937af --- /dev/null +++ b/internal/audiobooks/abs/revocation_test.go @@ -0,0 +1,204 @@ +package abs + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/streamrevoke" + "github.com/go-chi/chi/v5" + "github.com/golang-jwt/jwt/v5" +) + +func TestBearerAuthCarriesJWTTimestampForUserCutoff(t *testing.T) { + const accessTokenType = "access" + secret := []byte("test-secret-32-bytes-aaaaaaaaaaaaa") + tokens := newMemTokenStore() + revocations := streamrevoke.New(streamrevoke.Options{}) + if err := revocations.RevokeUser(context.Background(), 7, "cutoff"); err != nil { + t.Fatal(err) + } + cutoff := revocations.List()[0].RevokedAt + h := New(Dependencies{ + MediaStore: noopMediaStore{}, + TokenStore: tokens, + Config: &staticConfig{secret: secret}, + Revocation: revocations, + }) + + issue := func(jti string, issuedAt time.Time) string { + t.Helper() + raw, err := issueJWT(secret, Claims{ + Type: accessTokenType, + UserID: "7", + JTI: jti, + RegisteredClaims: jwt.RegisteredClaims{ + IssuedAt: jwt.NewNumericDate(issuedAt), + ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour)), + }, + }) + if err != nil { + t.Fatal(err) + } + if err := tokens.InsertToken(context.Background(), ABSToken{ + UserID: "7", Type: accessTokenType, JTI: jti, ExpiresAt: time.Now().Add(time.Hour), + }); err != nil { + t.Fatal(err) + } + return raw + } + + next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + auth, ok := absAuthFrom(r) + if !ok { + t.Fatal("missing ctxAuth") + } + if revocations.Refuse(w, "", 7, auth.IssuedAt) { + return + } + w.WriteHeader(http.StatusNoContent) + }) + protected := h.bearerAuth(next) + for _, tc := range []struct { + name string + token string + status int + }{ + {name: "before cutoff refused", token: issue("old", cutoff.Add(-time.Hour)), status: http.StatusForbidden}, + {name: "after cutoff admitted", token: issue("new", cutoff.Add(time.Hour)), status: http.StatusNoContent}, + } { + t.Run(tc.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.Header.Set("Authorization", "Bearer "+tc.token) + rec := httptest.NewRecorder() + protected.ServeHTTP(rec, req) + if rec.Code != tc.status { + t.Fatalf("status=%d body=%s, want %d", rec.Code, rec.Body.String(), tc.status) + } + }) + } +} + +func TestPublicTrackUsesSessionStartedAtForCutoff(t *testing.T) { + h, _ := newPublicTrackHandler(t, "sid-cutoff", "book-1", false) + store := streamrevoke.New(streamrevoke.Options{}) + h.deps.Revocation = store + if err := store.RevokeUser(context.Background(), 1, "cutoff"); err != nil { + t.Fatal(err) + } + + rec := dispatchTrack(h, http.MethodGet, "sid-cutoff", "1") + if rec.Code != http.StatusForbidden { + t.Fatalf("status=%d body=%s, want 403", rec.Code, rec.Body.String()) + } + + sessions, ok := h.deps.PlaybackSessionStore.(*fakePlaybackSessionStore) + if !ok { + t.Fatal("unexpected playback session store type") + } + session := sessions.sessions["sid-cutoff"] + session.StartedAt = time.Now().Add(time.Hour) + sessions.sessions["sid-cutoff"] = session + rec = dispatchTrack(h, http.MethodGet, "sid-cutoff", "1") + if rec.Code != http.StatusOK { + t.Fatalf("post-cutoff session status=%d body=%s, want 200", rec.Code, rec.Body.String()) + } +} + +type revocationFeedFileMediaStore struct { + noopMediaStore + file *models.MediaFile + onGet func() +} + +func (s *revocationFeedFileMediaStore) GetMediaFileByID(_ context.Context, id int) (*models.MediaFile, error) { + if s.onGet != nil { + s.onGet() + } + if s.file != nil && s.file.ID == id { + return s.file, nil + } + return nil, ErrNotFound +} + +type deadlineResponseRecorder struct { + *httptest.ResponseRecorder + deadlines int +} + +func (w *deadlineResponseRecorder) SetWriteDeadline(time.Time) error { + w.deadlines++ + return nil +} + +func TestPublicFeedFileUsesFeedCreatedAtForCutoff(t *testing.T) { + const ( + feedSlug = "cutoff-feed" + ino = 9 + ) + ctx := context.Background() + feeds := newMemRSSFeedStore() + feed := RSSFeed{ + ID: "feed-1", + UserID: "1", + LibraryItemID: "book-1", + Slug: feedSlug, + CreatedAt: time.Now().Add(-time.Hour), + } + feeds.rows[feed.ID] = feed + revocations := streamrevoke.New(streamrevoke.Options{}) + if err := revocations.RevokeUser(ctx, 1, "cutoff"); err != nil { + t.Fatal(err) + } + cutoff := revocations.List()[0].RevokedAt + mediaStore := &revocationFeedFileMediaStore{file: &models.MediaFile{ + ID: ino, ContentID: feed.LibraryItemID, FilePath: makeTempAudio(t), + }} + h := New(Dependencies{ + MediaStore: mediaStore, + RSSFeedStore: feeds, + Revocation: revocations, + }) + + rec := dispatchABSWithParams(http.MethodGet, "/feed/"+feedSlug+"/file/9", + map[string]string{"slug": feedSlug, "ino": "9"}, nil, "", "", h.handlePublicFeedFile) + if rec.Code != http.StatusForbidden { + t.Fatalf("pre-cutoff feed status=%d body=%s, want 403", rec.Code, rec.Body.String()) + } + + feed.CreatedAt = cutoff.Add(time.Nanosecond) + feeds.rows[feed.ID] = feed + for !time.Now().After(feed.CreatedAt) { + time.Sleep(time.Nanosecond) + } + rec = dispatchABSWithParams(http.MethodGet, "/feed/"+feedSlug+"/file/9", + map[string]string{"slug": feedSlug, "ino": "9"}, nil, "", "", h.handlePublicFeedFile) + if rec.Code != http.StatusOK { + t.Fatalf("post-cutoff feed status=%d body=%s, want 200", rec.Code, rec.Body.String()) + } + + // Revoke again between the handler's initial Refuse check and WatchAndCut. + // The deadline recorder makes the watcher observable without a real socket. + mediaStore.onGet = func() { + mediaStore.onGet = nil + if err := revocations.RevokeUser(ctx, 1, "cut during pour"); err != nil { + t.Fatalf("RevokeUser during pour: %v", err) + } + } + req := httptest.NewRequest(http.MethodGet, "/feed/"+feedSlug+"/file/9", nil) + rctx := chi.NewRouteContext() + rctx.URLParams.Add("slug", feedSlug) + rctx.URLParams.Add("ino", "9") + req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, rctx)) + deadlineRec := &deadlineResponseRecorder{ResponseRecorder: httptest.NewRecorder()} + h.handlePublicFeedFile(deadlineRec, req) + if deadlineRec.Code != http.StatusOK { + t.Fatalf("watched feed status=%d body=%s, want 200", deadlineRec.Code, deadlineRec.Body.String()) + } + if deadlineRec.deadlines != 1 { + t.Fatalf("write deadlines=%d, want 1 from WatchAndCut", deadlineRec.deadlines) + } +} diff --git a/internal/audiobooks/abs/rss_feeds_handler.go b/internal/audiobooks/abs/rss_feeds_handler.go index 47678dcc..c8a49f3e 100644 --- a/internal/audiobooks/abs/rss_feeds_handler.go +++ b/internal/audiobooks/abs/rss_feeds_handler.go @@ -224,6 +224,14 @@ func (h *Handler) handlePublicFeedFile(w http.ResponseWriter, r *http.Request) { http.Error(w, "feed get failed", http.StatusInternalServerError) return } + userID, parseUserErr := strconv.Atoi(f.UserID) + if parseUserErr != nil || userID <= 0 { + http.Error(w, "feed not found", http.StatusNotFound) + return + } + if h.deps.Revocation != nil && h.deps.Revocation.Refuse(w, "", userID, f.CreatedAt) { + return + } inoStr := chi.URLParam(r, "ino") ino, parseErr := strconv.Atoi(inoStr) if parseErr != nil { @@ -235,5 +243,7 @@ func (h *Handler) handlePublicFeedFile(w http.ResponseWriter, r *http.Request) { http.Error(w, "file not found", http.StatusNotFound) return } + stop := h.deps.Revocation.WatchAndCut(w, "", userID, f.CreatedAt) + defer stop() http.ServeFile(w, r, mf.FilePath) } diff --git a/internal/audiobooks/service.go b/internal/audiobooks/service.go index faef425b..d8a52b2f 100644 --- a/internal/audiobooks/service.go +++ b/internal/audiobooks/service.go @@ -12,6 +12,7 @@ import ( "github.com/Silo-Server/silo-server/internal/playback" "github.com/Silo-Server/silo-server/internal/recommendations" "github.com/Silo-Server/silo-server/internal/scanner" + "github.com/Silo-Server/silo-server/internal/streamrevoke" "github.com/jackc/pgx/v5/pgxpool" ) @@ -60,6 +61,7 @@ type ABSHandlerDeps struct { Detail *catalog.DetailService SessionMgr *playback.SessionManager SessionSyncer abs.PlaybackSessionSyncer + Revocation *streamrevoke.Store } // absAuthAdapter is the narrow slice of internal/auth that BuildABSHandler @@ -174,6 +176,7 @@ func (s *Service) BuildABSHandler(deps ABSHandlerDeps) *abs.Handler { SocketIO: socketServer, NativeSessions: deps.SessionMgr, NativeSessionSyncer: deps.SessionSyncer, + Revocation: deps.Revocation, CoverResolver: func(ctx context.Context, path, variant string) string { if deps.Detail == nil { return ""