From fca9f38a330566389bee4e43772aec7f2646aa7a Mon Sep 17 00:00:00 2001 From: CoffeeKnyte <67730400+CoffeeKnyte@users.noreply.github.com> Date: Wed, 29 Jul 2026 15:17:15 +0000 Subject: [PATCH] feat(api): operator-visible stream kill list with unrevoke MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The kill switch had no operator surface. streamrevoke.Store.List() existed and was called from nowhere, so an admin could only terminate one session by id or revoke a user's streams as a side effect of editing their account — with no way to see what was revoked, choose a TTL, or undo a mistake. Because expiry is deliberately monotonic, a wrong 24h kill was irreversible. - GET/POST/DELETE /api/v1/admin/streams/revocations, admin-only, additive. - Explicit wire-to-internal kind mapping: the wire accepts "session" (and "sess"), the store key is "sess". Passing the wire string straight into a Key would create a revocation IsRevoked never consults. - Validation: non-empty bounded session ids; canonical positive user ids (strconv.Itoa round-trip, so "01" is rejected — the cache key is the canonical form); ttl_seconds bounded to 30d so the duration cannot overflow; bounded reason and request body. DELETE of an absent key is idempotent. - Store.Unrevoke, guarded by a bounded in-memory tombstone: the tombstone is installed and the local entry dropped BEFORE the slow durable/Redis deletes, so a concurrent poll reconcile cannot re-apply the row it just read. A newer Revoke on the same key clears the tombstone, so an unrevoke never suppresses a later legitimate kill. Tombstones age out with the kill they replaced. - maintain() takes the same operation lock as Revoke/Unrevoke around its durable block, closing the window where a poll tick could re-Upsert a row Unrevoke had just deleted — invisible until a restart resurrected the kill. - Durable self-heal now compares expiry, not mere presence, so a failed Revoke mirror leaves a stale shorter row that the next tick repairs. - A failed unrevoke publish fails safe: other processes keep the kill until it expires. Propagation failures surface as warnings rather than weakening the local result. Unrevoking an over-cap victim is legal but the async enforcer will re-revoke it on its next pass while the user is still over cap; that is documented at the endpoint. Part of the stream monitoring & kill-switch epic. --- .../api/handlers/admin_stream_revocations.go | 178 +++++++++++++ .../handlers/admin_stream_revocations_test.go | 124 +++++++++ internal/api/router.go | 6 + internal/streamrevoke/durable_postgres.go | 8 + internal/streamrevoke/store.go | 242 +++++++++++++++--- internal/streamrevoke/store_test.go | 200 ++++++++++++++- 6 files changed, 719 insertions(+), 39 deletions(-) create mode 100644 internal/api/handlers/admin_stream_revocations.go create mode 100644 internal/api/handlers/admin_stream_revocations_test.go diff --git a/internal/api/handlers/admin_stream_revocations.go b/internal/api/handlers/admin_stream_revocations.go new file mode 100644 index 00000000..94018418 --- /dev/null +++ b/internal/api/handlers/admin_stream_revocations.go @@ -0,0 +1,178 @@ +package handlers + +import ( + "encoding/json" + "errors" + "io" + "net/http" + "sort" + "strconv" + "strings" + "time" + + "github.com/Silo-Server/silo-server/internal/streamrevoke" + "github.com/go-chi/chi/v5" +) + +const ( + maxStreamRevocationBodyBytes = 16 << 10 + maxStreamRevocationIDLength = 256 + maxStreamRevocationReason = 512 + maxStreamRevocationTTL = 30 * 24 * 60 * 60 +) + +// AdminStreamRevocationHandler exposes the operator-visible stream kill list. +// Unrevoking an over-cap victim is legal, but the async enforcer may revoke it +// again on its next pass while the owning user remains over cap. +type AdminStreamRevocationHandler struct { + store *streamrevoke.Store +} + +func NewAdminStreamRevocationHandler(store *streamrevoke.Store) *AdminStreamRevocationHandler { + return &AdminStreamRevocationHandler{store: store} +} + +type streamRevocationRequest struct { + Kind string `json:"kind"` + ID string `json:"id"` + Reason string `json:"reason"` + TTLSeconds *int64 `json:"ttl_seconds"` +} + +type streamRevocationResponse struct { + streamrevoke.Revocation + Warnings []string `json:"warnings,omitempty"` +} + +func (h *AdminStreamRevocationHandler) HandleList(w http.ResponseWriter, _ *http.Request) { + revocations := h.store.List() + sort.Slice(revocations, func(i, j int) bool { + return revocations[i].RevokedAt.After(revocations[j].RevokedAt) + }) + writeJSON(w, http.StatusOK, map[string]any{"revocations": revocations}) +} + +func (h *AdminStreamRevocationHandler) HandleCreate(w http.ResponseWriter, r *http.Request) { + var req streamRevocationRequest + r.Body = http.MaxBytesReader(w, r.Body, maxStreamRevocationBodyBytes) + decoder := json.NewDecoder(r.Body) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body") + return + } + if err := ensureJSONEOF(decoder); err != nil { + writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body") + return + } + + key, err := revocationKeyFromWire(req.Kind, req.ID) + if err != nil { + writeError(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + reason := strings.TrimSpace(req.Reason) + if len(reason) > maxStreamRevocationReason { + writeError(w, http.StatusBadRequest, "bad_request", "Reason is too long") + return + } + + var until time.Time + if req.TTLSeconds != nil { + if *req.TTLSeconds <= 0 || *req.TTLSeconds > maxStreamRevocationTTL { + writeError(w, http.StatusBadRequest, "bad_request", "ttl_seconds must be between 1 and 2592000") + return + } + until = time.Now().Add(time.Duration(*req.TTLSeconds) * time.Second) + } else { + // A zero expiry asks the store to use its default via the public helper. + var warnings []string + switch key.Kind { + case streamrevoke.KindSession: + warnings, err = h.store.RevokeSessionWithWarnings(r.Context(), key.ID, reason) + case streamrevoke.KindUser: + userID, _ := strconv.Atoi(key.ID) + warnings, err = h.store.RevokeUserWithWarnings(r.Context(), userID, reason) + } + if err != nil { + writeError(w, http.StatusInternalServerError, "internal_error", "Failed to create revocation") + return + } + writeJSON(w, http.StatusCreated, streamRevocationResponse{Revocation: findRevocation(h.store, key), Warnings: warnings}) + return + } + + warnings, err := h.store.RevokeWithWarnings(r.Context(), key, reason, until) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal_error", "Failed to create revocation") + return + } + writeJSON(w, http.StatusCreated, streamRevocationResponse{Revocation: findRevocation(h.store, key), Warnings: warnings}) +} + +func (h *AdminStreamRevocationHandler) HandleDelete(w http.ResponseWriter, r *http.Request) { + key, err := revocationKeyFromWire(chi.URLParam(r, "kind"), chi.URLParam(r, "id")) + if err != nil { + writeError(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + warnings, err := h.store.UnrevokeWithWarnings(r.Context(), key) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal_error", "Failed to remove revocation") + return + } + if len(warnings) > 0 { + writeJSON(w, http.StatusOK, map[string]any{"warnings": warnings}) + return + } + w.WriteHeader(http.StatusNoContent) +} + +func revocationKeyFromWire(kind, id string) (streamrevoke.Key, error) { + var internalKind streamrevoke.Kind + switch kind { + case "session", string(streamrevoke.KindSession): + internalKind = streamrevoke.KindSession + case string(streamrevoke.KindUser): + internalKind = streamrevoke.KindUser + default: + return streamrevoke.Key{}, errors.New("kind must be session, sess, or user") + } + + if internalKind == streamrevoke.KindSession { + id = strings.TrimSpace(id) + if id == "" { + return streamrevoke.Key{}, errors.New("session id is required") + } + if len(id) > maxStreamRevocationIDLength { + return streamrevoke.Key{}, errors.New("session id is too long") + } + return streamrevoke.Key{Kind: internalKind, ID: id}, nil + } + + userID, err := strconv.Atoi(id) + if err != nil || userID <= 0 || strconv.Itoa(userID) != id { + return streamrevoke.Key{}, errors.New("user id must be a canonical positive integer") + } + return streamrevoke.Key{Kind: internalKind, ID: id}, nil +} + +func ensureJSONEOF(decoder *json.Decoder) error { + var extra any + if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) { + if err == nil { + return errors.New("multiple JSON values") + } + return err + } + return nil +} + +func findRevocation(store *streamrevoke.Store, key streamrevoke.Key) streamrevoke.Revocation { + for _, revocation := range store.List() { + if revocation.Kind == key.Kind && revocation.ID == key.ID { + return revocation + } + } + return streamrevoke.Revocation{Kind: key.Kind, ID: key.ID} +} diff --git a/internal/api/handlers/admin_stream_revocations_test.go b/internal/api/handlers/admin_stream_revocations_test.go new file mode 100644 index 00000000..52057840 --- /dev/null +++ b/internal/api/handlers/admin_stream_revocations_test.go @@ -0,0 +1,124 @@ +package handlers + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + apimw "github.com/Silo-Server/silo-server/internal/api/middleware" + "github.com/Silo-Server/silo-server/internal/auth" + "github.com/Silo-Server/silo-server/internal/streamrevoke" + "github.com/go-chi/chi/v5" +) + +func streamRevocationTestRouter(store *streamrevoke.Store) http.Handler { + h := NewAdminStreamRevocationHandler(store) + r := chi.NewRouter() + r.Get("/revocations", h.HandleList) + r.Post("/revocations", h.HandleCreate) + r.Delete("/revocations/{kind}/{id}", h.HandleDelete) + return r +} + +func TestAdminStreamRevocationRequiresAdminMiddleware(t *testing.T) { + const adminRole = "admin" + protected := apimw.RequireActingAdmin(nil)(streamRevocationTestRouter(streamrevoke.New(streamrevoke.Options{}))) + for _, tc := range []struct { + name string + claims *auth.Claims + status int + }{ + {name: "unauthenticated", status: http.StatusUnauthorized}, + {name: "non admin", claims: &auth.Claims{UserID: 1, Role: string(scopeUser)}, status: http.StatusForbidden}, + {name: adminRole, claims: &auth.Claims{UserID: 1, Role: adminRole}, status: http.StatusOK}, + } { + t.Run(tc.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/revocations", nil) + if tc.claims != nil { + req = req.WithContext(apimw.SetClaims(req.Context(), tc.claims)) + } + rec := httptest.NewRecorder() + protected.ServeHTTP(rec, req) + if rec.Code != tc.status { + t.Fatalf("status=%d, want %d", rec.Code, tc.status) + } + }) + } +} + +func TestAdminStreamRevocationSessionMappingAndListDeleteRoundTrip(t *testing.T) { + store := streamrevoke.New(streamrevoke.Options{}) + router := streamRevocationTestRouter(store) + req := httptest.NewRequest(http.MethodPost, "/revocations", bytes.NewBufferString(`{"kind":"session","id":" s1 ","reason":" test "}`)) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("create status=%d body=%s", rec.Code, rec.Body.String()) + } + var created map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &created); err != nil { + t.Fatal(err) + } + if created["kind"] != "sess" || created["id"] != "s1" { + t.Fatalf("created=%v, want internal sess kind and trimmed id", created) + } + + listReq := httptest.NewRequest(http.MethodGet, "/revocations", nil) + listRec := httptest.NewRecorder() + router.ServeHTTP(listRec, listReq) + var listed struct { + Revocations []streamrevoke.Revocation `json:"revocations"` + } + if err := json.Unmarshal(listRec.Body.Bytes(), &listed); err != nil { + t.Fatal(err) + } + if len(listed.Revocations) != 1 { + t.Fatalf("list=%+v", listed.Revocations) + } + deletePath := "/revocations/" + string(listed.Revocations[0].Kind) + "/" + listed.Revocations[0].ID + deleteReq := httptest.NewRequest(http.MethodDelete, deletePath, nil) + deleteRec := httptest.NewRecorder() + router.ServeHTTP(deleteRec, deleteReq) + if deleteRec.Code != http.StatusNoContent || len(store.List()) != 0 { + t.Fatalf("delete status=%d remaining=%v", deleteRec.Code, store.List()) + } +} + +func TestAdminStreamRevocationValidation(t *testing.T) { + router := streamRevocationTestRouter(streamrevoke.New(streamrevoke.Options{})) + tests := []struct { + name string + body string + }{ + {name: "noncanonical user", body: `{"kind":"user","id":"01"}`}, + {name: "zero user", body: `{"kind":"user","id":"0"}`}, + {name: "unknown kind", body: `{"kind":"stream","id":"x"}`}, + {name: "missing session id", body: `{"kind":"session","id":" "}`}, + {name: "zero ttl", body: `{"kind":"sess","id":"x","ttl_seconds":0}`}, + {name: "negative ttl", body: `{"kind":"sess","id":"x","ttl_seconds":-1}`}, + {name: "ttl above max", body: `{"kind":"sess","id":"x","ttl_seconds":2592001}`}, + {name: "ttl integer overflow", body: `{"kind":"sess","id":"x","ttl_seconds":9223372036854775808}`}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/revocations", bytes.NewBufferString(tt.body)) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String()) + } + }) + } +} + +func TestAdminStreamRevocationDeleteAbsentIsIdempotent(t *testing.T) { + router := streamRevocationTestRouter(streamrevoke.New(streamrevoke.Options{})) + req := httptest.NewRequest(http.MethodDelete, "/revocations/sess/missing", nil) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusNoContent { + t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String()) + } +} diff --git a/internal/api/router.go b/internal/api/router.go index a420f1bf..0222ce2d 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -3103,6 +3103,12 @@ func NewRouter(deps Dependencies) chi.Router { r.Post("/sessions/{session_id}/terminate", adminPlaybackControlHandler.HandleTerminateSession) r.Post("/sessions/{session_id}/message", adminPlaybackControlHandler.HandleMessageSession) } + if deps.RevocationStore != nil { + streamRevocationHandler := handlers.NewAdminStreamRevocationHandler(deps.RevocationStore) + r.Get("/streams/revocations", streamRevocationHandler.HandleList) + r.Post("/streams/revocations", streamRevocationHandler.HandleCreate) + r.Delete("/streams/revocations/{kind}/{id}", streamRevocationHandler.HandleDelete) + } if deps.TaskManager != nil { taskHistoryRepo := repository.NewPgExecutionRepository(deps.DB) diff --git a/internal/streamrevoke/durable_postgres.go b/internal/streamrevoke/durable_postgres.go index 99c3d73a..1e9c3906 100644 --- a/internal/streamrevoke/durable_postgres.go +++ b/internal/streamrevoke/durable_postgres.go @@ -69,6 +69,14 @@ func (s *PostgresDurableStore) Upsert(ctx context.Context, r Revocation) error { return nil } +// Delete removes a revocation from the durable mirror. +func (s *PostgresDurableStore) Delete(ctx context.Context, kind Kind, id string) error { + if _, err := s.pool.Exec(ctx, `DELETE FROM stream_revocations WHERE kind = $1 AND id = $2`, string(kind), id); err != nil { + return fmt.Errorf("streamrevoke delete: %w", err) + } + return nil +} + // ListActive returns every revocation not yet expired, for warming the hot-path // cache on startup and re-warming after a Redis flush. func (s *PostgresDurableStore) ListActive(ctx context.Context) ([]Revocation, error) { diff --git a/internal/streamrevoke/store.go b/internal/streamrevoke/store.go index b8f08867..1864e24e 100644 --- a/internal/streamrevoke/store.go +++ b/internal/streamrevoke/store.go @@ -36,6 +36,8 @@ const ( // EventStreamRevoked is the cache.Event.Type published on cache.ChannelAdmin // when a revocation is created so other nodes apply it immediately. EventStreamRevoked = "stream_revoked" + // EventStreamUnrevoked is published when an operator removes a revocation. + EventStreamUnrevoked = "stream_unrevoked" defaultPollInterval = 60 * time.Second // defaultTTL is intentionally >= the playback recipe-card lifetime @@ -87,6 +89,7 @@ func (r Revocation) expired(now time.Time) bool { // the interface when one is provided. type DurableStore interface { Upsert(ctx context.Context, r Revocation) error + Delete(ctx context.Context, kind Kind, id string) error ListActive(ctx context.Context) ([]Revocation, error) Prune(ctx context.Context) error } @@ -108,8 +111,11 @@ type Store struct { pollInterval time.Duration defaultTTL time.Duration - mu sync.RWMutex - items map[Key]Revocation + opMu sync.Mutex + mu sync.RWMutex + items map[Key]Revocation + tombstones map[Key]time.Time + tombstoneExpires map[Key]time.Time } // New builds a Store, applying defaults for any unset Options. @@ -121,12 +127,14 @@ func New(opts Options) *Store { opts.DefaultTTL = defaultTTL } return &Store{ - rdb: opts.Redis, - bus: opts.Bus, - durable: opts.Durable, - pollInterval: opts.PollInterval, - defaultTTL: opts.DefaultTTL, - items: make(map[Key]Revocation), + rdb: opts.Redis, + bus: opts.Bus, + durable: opts.Durable, + pollInterval: opts.PollInterval, + defaultTTL: opts.DefaultTTL, + items: make(map[Key]Revocation), + tombstones: make(map[Key]time.Time), + tombstoneExpires: make(map[Key]time.Time), } } @@ -207,14 +215,15 @@ func (s *Store) WatchAndCut(w http.ResponseWriter, sessionID string, userID int, // // startedAt is when the request's stream credential was issued: the stream // token's iat on token-bearing surfaces (edge proxy, transcode node, native -// ?st=), or the request entry time on freshly-authenticated surfaces (native -// session auth, jellycompat login). A user revocation is a CUTOFF, not a ban: +// ?st=, ABS bearer), the persisted playback-session start for public ABS +// tracks, or the feed creation time for public RSS capabilities. A user +// revocation is a CUTOFF, not a ban: // it kills streams whose credential predates it, while a stream authorized // after it (which required passing auth that the revocation just reset) plays // normally. Without the cutoff, the OnUserSessionsRevoked hook — fired by any // admin edit of password/role/enabled/permissions/quality — would 403 the -// user's playback for the full 24h TTL even after they re-authenticate, with -// no unrevoke path (expiry is deliberately monotonic). A zero startedAt never +// user's playback for the full 24h TTL even after they re-authenticate unless +// an operator explicitly unrevoked it. A zero startedAt never // matches a user revocation (fail open, matching the enforcer's "never kill on // uncertainty"); session revocations are exact-id kills and ignore startedAt. func (s *Store) IsRevoked(sessionID string, userID int, startedAt time.Time) bool { @@ -258,6 +267,14 @@ func (r Revocation) effectiveExpiry() time.Time { // not fight a freshly-applied one. The caller must not hold s.mu. func (s *Store) applyLocal(r Revocation) { s.mu.Lock() + if issuedAt, ok := s.tombstones[r.key()]; ok { + if !r.RevokedAt.After(issuedAt) { + s.mu.Unlock() + return + } + delete(s.tombstones, r.key()) + delete(s.tombstoneExpires, r.key()) + } if existing, ok := s.items[r.key()]; !ok || !existing.effectiveExpiry().After(r.effectiveExpiry()) { s.items[r.key()] = r } @@ -273,11 +290,29 @@ func (s *Store) effective(k Key) Revocation { return s.items[k] } +func (s *Store) isTombstoned(k Key) bool { + s.mu.RLock() + defer s.mu.RUnlock() + _, ok := s.tombstones[k] + return ok +} + // Revoke adds a revocation to the local cache, writes it to Redis (with a TTL // of until-now) when configured, mirrors it to the durable store when // configured, and publishes a pub/sub event so other nodes apply it // immediately. func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.Time) error { + _, err := s.RevokeWithWarnings(ctx, key, reason, until) + return err +} + +// RevokeWithWarnings applies a revocation locally and reports propagation +// failures without weakening the local kill. Revoke is the compatibility +// wrapper used by existing callers that only need local-success semantics. +func (s *Store) RevokeWithWarnings(ctx context.Context, key Key, reason string, until time.Time) ([]string, error) { + s.opMu.Lock() + defer s.opMu.Unlock() + now := time.Now() r := Revocation{ Kind: key.Kind, @@ -290,6 +325,7 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T // Local cache first: the hot path must reflect the kill immediately even if // downstream propagation fails. s.applyLocal(r) + var warnings []string // Propagation must not die with the caller: an admin terminate rides its // HTTP request's context, and an abort after applyLocal would strand the @@ -301,6 +337,7 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T if s.durable != nil { if err := s.durable.Upsert(ctx, r); err != nil { slog.Warn("streamrevoke durable upsert failed", "error", err, "kind", r.Kind, "id", r.ID) + warnings = append(warnings, "durable mirror was not updated") } } @@ -308,7 +345,9 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T // applyLocal kept (monotonic): a short over-cap re-revoke must not shorten a // longer admin kill's Redis TTL, matching applyLocal and the durable upsert's // GREATEST. Edges that warm from Redis in that window must see the later kill. - s.mirrorToRedis(ctx, s.effective(r.key())) + if err := s.mirrorToRedis(ctx, s.effective(r.key())); err != nil { + warnings = append(warnings, "Redis mirror was not updated") + } // Pub/sub push so already-connected nodes apply the kill immediately (the // warm loops deliberately skip this — SCAN reconcile handles late-joiners). @@ -316,15 +355,16 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T data, err := json.Marshal(r) if err != nil { slog.Warn("streamrevoke marshal failed", "error", err, "kind", r.Kind, "id", r.ID) - return err + return warnings, err } evt := cache.Event{Type: EventStreamRevoked, Payload: string(data)} if err := s.bus.Publish(ctx, cache.ChannelAdmin, evt); err != nil { slog.Warn("streamrevoke publish failed", "error", err, "kind", r.Kind, "id", r.ID) + warnings = append(warnings, "revocation event was not published") } } - return nil + return warnings, nil } // mirrorToRedis writes (or deletes) the Redis mirror key for a revocation with a @@ -334,14 +374,14 @@ func (s *Store) Revoke(ctx context.Context, key Key, reason string, until time.T // Redis, so re-arming Redis from the durable source of truth after a Redis flush // + central restart is what lets edges reconverge via their SCAN reconcile. // Best-effort: failures are logged, never returned. No-op when Redis is absent. -func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) { +func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) error { if s.rdb == nil { - return + return nil } data, err := json.Marshal(r) if err != nil { slog.Warn("streamrevoke marshal failed", "error", err, "kind", r.Kind, "id", r.ID) - return + return err } if r.ExpiresAt.IsZero() { // Permanent kill: set without a TTL, matching applyLocal/effectiveExpiry @@ -349,20 +389,105 @@ func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) { // hugely negative and drop the key, so late-joining edges would miss it. if err := s.rdb.Set(ctx, redisKey(r.key()), data, 0).Err(); err != nil { slog.Warn("streamrevoke redis set failed", "error", err, "kind", r.Kind, "id", r.ID) + return err } - return + return nil } ttl := time.Until(r.ExpiresAt) if ttl <= 0 { // Already expired: nothing to mirror in Redis. if err := s.rdb.Del(ctx, redisKey(r.key())).Err(); err != nil { slog.Debug("streamrevoke redis del failed", "error", err, "kind", r.Kind, "id", r.ID) + return err } - return + return nil } if err := s.rdb.Set(ctx, redisKey(r.key()), data, ttl).Err(); err != nil { slog.Warn("streamrevoke redis set failed", "error", err, "kind", r.Kind, "id", r.ID) + return err } + return nil +} + +type unrevocation struct { + Kind Kind `json:"kind"` + ID string `json:"id"` + IssuedAt time.Time `json:"issued_at"` + ExpiresAt time.Time `json:"expires_at"` +} + +func (u unrevocation) key() Key { + return Key{Kind: u.Kind, ID: u.ID} +} + +// applyUnrevocation installs a bounded tombstone before deleting the local +// kill. A delayed unrevoke event never removes a newer revocation. +func (s *Store) applyUnrevocation(u unrevocation) { + k := u.key() + s.mu.Lock() + if existing, ok := s.items[k]; ok && existing.RevokedAt.After(u.IssuedAt) { + s.mu.Unlock() + return + } + if previous, ok := s.tombstones[k]; ok && previous.After(u.IssuedAt) { + s.mu.Unlock() + return + } + expiresAt := u.ExpiresAt + maxExpiry := u.IssuedAt.Add(s.defaultTTL) + if expiresAt.IsZero() || expiresAt.After(maxExpiry) { + expiresAt = maxExpiry + } + s.tombstones[k] = u.IssuedAt + s.tombstoneExpires[k] = expiresAt + delete(s.items, k) + s.mu.Unlock() +} + +// Unrevoke removes a kill locally and from every configured mirror. A failed +// publish fails safe: other processes may retain the kill until expiry. +func (s *Store) Unrevoke(ctx context.Context, key Key) error { + _, err := s.UnrevokeWithWarnings(ctx, key) + return err +} + +// UnrevokeWithWarnings is the admin-facing form of Unrevoke. +func (s *Store) UnrevokeWithWarnings(ctx context.Context, key Key) ([]string, error) { + s.opMu.Lock() + defer s.opMu.Unlock() + + issuedAt := time.Now() + s.mu.RLock() + existing := s.items[key] + s.mu.RUnlock() + u := unrevocation{Kind: key.Kind, ID: key.ID, IssuedAt: issuedAt, ExpiresAt: existing.ExpiresAt} + s.applyUnrevocation(u) + + ctx = context.WithoutCancel(ctx) + var warnings []string + if s.durable != nil { + if err := s.durable.Delete(ctx, key.Kind, key.ID); err != nil { + slog.WarnContext(ctx, "streamrevoke durable delete failed", "error", err, "kind", key.Kind, "id", key.ID) + warnings = append(warnings, "durable mirror was not deleted") + } + } + if s.rdb != nil { + if err := s.rdb.Del(ctx, redisKey(key)).Err(); err != nil { + slog.WarnContext(ctx, "streamrevoke redis delete failed", "error", err, "kind", key.Kind, "id", key.ID) + warnings = append(warnings, "Redis mirror was not deleted") + } + } + if s.bus != nil { + data, err := json.Marshal(u) + if err != nil { + return warnings, err + } + if err := s.bus.Publish(ctx, cache.ChannelAdmin, cache.Event{Type: EventStreamUnrevoked, Payload: string(data)}); err != nil { + slog.WarnContext(ctx, "streamrevoke unrevoke publish failed", "error", err, "kind", key.Kind, "id", key.ID) + warnings = append(warnings, "unrevocation event was not published; other processes may retain the kill until it expires") + } + } + return warnings, nil } // RevokeSession revokes a single session id for DefaultTTL. @@ -370,6 +495,11 @@ func (s *Store) RevokeSession(ctx context.Context, sessionID, reason string) err return s.Revoke(ctx, Key{Kind: KindSession, ID: sessionID}, reason, time.Now().Add(s.defaultTTL)) } +// RevokeSessionWithWarnings is the admin-facing form of RevokeSession. +func (s *Store) RevokeSessionWithWarnings(ctx context.Context, sessionID, reason string) ([]string, error) { + return s.RevokeWithWarnings(ctx, Key{Kind: KindSession, ID: sessionID}, reason, time.Now().Add(s.defaultTTL)) +} + // RevokeSessionFor revokes a single session id for a caller-chosen TTL. The // async enforcer uses a short TTL so a transient over-count (e.g. a ghost // session lingering next to a fresh reconnect) self-heals; a persistent abuser @@ -398,6 +528,15 @@ func (s *Store) RevokeUser(ctx context.Context, userID int, reason string) error return s.Revoke(ctx, userKey(userID), reason, time.Now().Add(s.defaultTTL)) } +// RevokeUserWithWarnings applies the default TTL while preserving propagation +// warnings for the admin API. +func (s *Store) RevokeUserWithWarnings(ctx context.Context, userID int, reason string) ([]string, error) { + if userID <= 0 { + return nil, fmt.Errorf("streamrevoke: invalid userID %d", userID) + } + return s.RevokeWithWarnings(ctx, userKey(userID), reason, time.Now().Add(s.defaultTTL)) +} + // List returns the currently-active revocations, pruning expired entries on // read. func (s *Store) List() []Revocation { @@ -424,6 +563,12 @@ func (s *Store) pruneExpired() { delete(s.items, k) } } + for k, expiresAt := range s.tombstoneExpires { + if !now.Before(expiresAt) { + delete(s.tombstones, k) + delete(s.tombstoneExpires, k) + } + } s.mu.Unlock() } @@ -448,7 +593,9 @@ func (s *Store) StartSync(ctx context.Context) { for _, r := range revs { if !r.expired(now) { s.applyLocal(r) - s.mirrorToRedis(ctx, r) + if !s.isTombstoned(r.key()) { + _ = s.mirrorToRedis(ctx, s.effective(r.key())) + } } } } @@ -516,20 +663,43 @@ func (s *Store) maintain(ctx context.Context) { // flush self-heals, re-arming Redis for edge nodes too, then physically // reclaim expired rows. if s.durable != nil { + s.opMu.Lock() + if revs, err := s.durable.ListActive(ctx); err != nil { slog.Warn("streamrevoke durable reconcile failed", "error", err) } else { now := time.Now() + durable := make(map[Key]Revocation, len(revs)) for _, r := range revs { if !r.expired(now) { + durable[r.key()] = r s.applyLocal(r) - s.mirrorToRedis(ctx, r) + if !s.isTombstoned(r.key()) { + _ = s.mirrorToRedis(ctx, s.effective(r.key())) + } + } + } + s.mu.RLock() + local := make([]Revocation, 0, len(s.items)) + for k, r := range s.items { + if _, tombstoned := s.tombstones[k]; !tombstoned && !r.expired(now) { + local = append(local, r) + } + } + s.mu.RUnlock() + for _, r := range local { + durableCopy, ok := durable[r.key()] + if !ok || durableCopy.effectiveExpiry().Before(r.effectiveExpiry()) { + if err := s.durable.Upsert(ctx, r); err != nil { + slog.WarnContext(ctx, "streamrevoke durable self-heal failed", "error", err, "kind", r.Kind, "id", r.ID) + } } } } if err := s.durable.Prune(ctx); err != nil { slog.Warn("streamrevoke durable prune failed", "error", err) } + s.opMu.Unlock() } s.pruneExpired() } @@ -537,18 +707,24 @@ func (s *Store) maintain(ctx context.Context) { // handleEvent applies an EventStreamRevoked event to the local cache. Other // event types on the channel are ignored. func (s *Store) handleEvent(evt cache.Event) { - if evt.Type != EventStreamRevoked { - return + switch evt.Type { + case EventStreamRevoked: + var r Revocation + if err := json.Unmarshal([]byte(evt.Payload), &r); err != nil { + slog.Debug("streamrevoke event unmarshal failed", "error", err) + return + } + if !r.expired(time.Now()) { + s.applyLocal(r) + } + case EventStreamUnrevoked: + var u unrevocation + if err := json.Unmarshal([]byte(evt.Payload), &u); err != nil { + slog.Debug("streamrevoke unrevoke event unmarshal failed", "error", err) + return + } + s.applyUnrevocation(u) } - var r Revocation - if err := json.Unmarshal([]byte(evt.Payload), &r); err != nil { - slog.Debug("streamrevoke event unmarshal failed", "error", err) - return - } - if r.expired(time.Now()) { - return - } - s.applyLocal(r) } // reconcileFromRedis SCANs the revocation namespace and applies every live diff --git a/internal/streamrevoke/store_test.go b/internal/streamrevoke/store_test.go index 31ab4b43..664665d8 100644 --- a/internal/streamrevoke/store_test.go +++ b/internal/streamrevoke/store_test.go @@ -2,10 +2,14 @@ package streamrevoke import ( "context" + "encoding/json" "errors" "sync" "testing" "time" + + "github.com/Silo-Server/silo-server/internal/cache" + "github.com/redis/go-redis/v9" ) // errFakeDurable is the error fakeDurable.ListActive returns while failNext > 0. @@ -23,12 +27,31 @@ func newMemStore() *Store { // mutex-guarded. failNext, when > 0, makes ListActive return an error that many // times before succeeding, to exercise the bounded boot-warm retry. type fakeDurable struct { - mu sync.Mutex - rows map[Key]Revocation - upserts int - prunes int - failNext int - lists int + mu sync.Mutex + rows map[Key]Revocation + upserts int + prunes int + deletes int + failNext int + lists int + upsertStarted chan struct{} + unblockUpsert chan struct{} +} + +type failingBus struct{} + +func (failingBus) Publish(context.Context, string, cache.Event) error { + return errors.New("publish failed") +} +func (failingBus) Subscribe(context.Context, string, cache.EventHandler) error { return nil } +func (failingBus) Close() error { return nil } + +func (f *fakeDurable) Delete(_ context.Context, kind Kind, id string) error { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.rows, Key{Kind: kind, ID: id}) + f.deletes++ + return nil } func newFakeDurable() *fakeDurable { @@ -36,6 +59,10 @@ func newFakeDurable() *fakeDurable { } func (f *fakeDurable) Upsert(_ context.Context, r Revocation) error { + if f.upsertStarted != nil { + close(f.upsertStarted) + <-f.unblockUpsert + } f.mu.Lock() defer f.mu.Unlock() f.rows[r.key()] = r @@ -221,6 +248,167 @@ func TestMaintainPrunesAndReWarmsFromDurable(t *testing.T) { } } +func TestMaintainDurableSelfHealDoesNotRaceUnrevoke(t *testing.T) { + ctx := context.Background() + fake := newFakeDurable() + fake.upsertStarted = make(chan struct{}) + fake.unblockUpsert = make(chan struct{}) + s := New(Options{Durable: fake}) + key := Key{Kind: KindSession, ID: "sess-maintain-unrevoke"} + s.applyLocal(Revocation{ + Kind: key.Kind, + ID: key.ID, + RevokedAt: time.Now(), + ExpiresAt: time.Now().Add(time.Hour), + }) + + maintainDone := make(chan struct{}) + go func() { + s.maintain(ctx) + close(maintainDone) + }() + <-fake.upsertStarted + + unrevokeDone := make(chan error, 1) + go func() { + unrevokeDone <- s.Unrevoke(ctx, key) + }() + + select { + case err := <-unrevokeDone: + t.Fatalf("Unrevoke completed before maintain released the operation lock: %v", err) + case <-time.After(20 * time.Millisecond): + } + + close(fake.unblockUpsert) + <-maintainDone + if err := <-unrevokeDone; err != nil { + t.Fatalf("Unrevoke: %v", err) + } + + fake.mu.Lock() + _, exists := fake.rows[key] + fake.mu.Unlock() + if exists { + t.Fatal("durable self-heal resurrected the unrevoked row") + } +} + +func TestUnrevokeTombstoneBlocksReconcileAndNewerRevokeClearsIt(t *testing.T) { + ctx := context.Background() + s := newMemStore() + key := Key{Kind: KindSession, ID: "sess-unrevoke"} + old := Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now().Add(-time.Minute), ExpiresAt: time.Now().Add(time.Hour)} + s.applyLocal(old) + + if err := s.Unrevoke(ctx, key); err != nil { + t.Fatalf("Unrevoke: %v", err) + } + s.applyLocal(old) + if s.IsRevoked(key.ID, 0, time.Time{}) { + t.Fatal("stale reconcile resurrected an unrevoked entry") + } + + if err := s.Revoke(ctx, key, "new kill", time.Now().Add(time.Hour)); err != nil { + t.Fatalf("newer Revoke: %v", err) + } + if !s.IsRevoked(key.ID, 0, time.Time{}) { + t.Fatal("newer revoke was suppressed by tombstone") + } + s.mu.RLock() + _, tombstoned := s.tombstones[key] + s.mu.RUnlock() + if tombstoned { + t.Fatal("newer revoke did not clear tombstone") + } +} + +func TestConcurrentRevokeUnrevokeThenReconcileDoesNotResurrect(t *testing.T) { + ctx := context.Background() + s := newMemStore() + key := Key{Kind: KindSession, ID: "sess-race"} + var wg sync.WaitGroup + for range 25 { + wg.Add(2) + go func() { + defer wg.Done() + _ = s.Revoke(ctx, key, "race", time.Now().Add(time.Hour)) + }() + go func() { + defer wg.Done() + _ = s.Unrevoke(ctx, key) + }() + } + wg.Wait() + + stale := Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now().Add(-time.Minute), ExpiresAt: time.Now().Add(time.Hour)} + if err := s.Unrevoke(ctx, key); err != nil { + t.Fatalf("final Unrevoke: %v", err) + } + s.applyLocal(stale) + if s.IsRevoked(key.ID, 0, time.Time{}) { + t.Fatal("stale reconcile resurrected entry after concurrent operations") + } +} + +func TestDelayedUnrevokeEventDoesNotDeleteNewerRevoke(t *testing.T) { + s := newMemStore() + key := Key{Kind: KindSession, ID: "sess-order"} + issuedAt := time.Now().Add(-time.Minute) + s.applyLocal(Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now(), ExpiresAt: time.Now().Add(time.Hour)}) + payload, err := json.Marshal(unrevocation{Kind: key.Kind, ID: key.ID, IssuedAt: issuedAt, ExpiresAt: time.Now().Add(time.Hour)}) + if err != nil { + t.Fatal(err) + } + s.handleEvent(cache.Event{Type: EventStreamUnrevoked, Payload: string(payload)}) + if !s.IsRevoked(key.ID, 0, time.Time{}) { + t.Fatal("delayed unrevoke event deleted a newer revocation") + } +} + +func TestUnrevokePropagationFailuresLeaveLocalStateUnrevoked(t *testing.T) { + rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:1", DialTimeout: 10 * time.Millisecond}) + t.Cleanup(func() { _ = rdb.Close() }) + s := New(Options{Redis: rdb, Bus: failingBus{}}) + key := Key{Kind: KindSession, ID: "sess-local"} + s.applyLocal(Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now().Add(-time.Minute), ExpiresAt: time.Now().Add(time.Hour)}) + + warnings, err := s.UnrevokeWithWarnings(context.Background(), key) + if err != nil { + t.Fatalf("UnrevokeWithWarnings: %v", err) + } + if len(warnings) != 2 { + t.Fatalf("warnings = %v, want Redis and publish failures", warnings) + } + if s.IsRevoked(key.ID, 0, time.Time{}) { + t.Fatal("local revocation remained after propagation failures") + } +} + +func TestMaintainRepairsShorterDurableExpiry(t *testing.T) { + ctx := context.Background() + fake := newFakeDurable() + key := Key{Kind: KindSession, ID: "sess-short-durable"} + short := Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now().Add(-time.Minute), ExpiresAt: time.Now().Add(time.Minute)} + long := Revocation{Kind: key.Kind, ID: key.ID, RevokedAt: time.Now(), ExpiresAt: time.Now().Add(time.Hour)} + fake.rows[key] = short + s := New(Options{Durable: fake}) + s.applyLocal(long) + + s.maintain(ctx) + + fake.mu.Lock() + got := fake.rows[key] + upserts := fake.upserts + fake.mu.Unlock() + if got.ExpiresAt.Before(long.ExpiresAt) { + t.Fatalf("durable expiry = %v, want at least %v", got.ExpiresAt, long.ExpiresAt) + } + if upserts == 0 { + t.Fatal("shorter durable row was not re-upserted") + } +} + func TestIsRevoked(t *testing.T) { ctx := context.Background()