From 90f5d117c8db01e21bf0f6aeb50412fd15379ec9 Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Sat, 1 Aug 2026 10:59:50 -0400 Subject: [PATCH] feat(events): let websocket clients declare channels on connect (#525) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(events): let websocket clients declare channels on connect Observing the events hub over `/api/v1/events/ws` costs more than it should. A read-only consumer has to send a `subscribe` frame within five seconds or be closed with a policy violation, which means implementing the handshake and holding the write half of the socket open purely to satisfy it. That cost is contract, not transport: `subscribe` is the only inbound message this endpoint accepts. Accept the selection on the URL instead. `?channels=catalog,user_state` subscribes on connect, answers with the same `subscribed` frame and per-channel snapshots the handshake produces, and is never put on the grace-period clock. A connection that declares nothing is unchanged — it still owes a subscribe frame within five seconds. Two supporting changes: - Channel selection now resolves through one shared function used by both paths, so the URL and handshake cannot drift on who may subscribe to what. Role, profile-binding, and validity checks are unchanged. - An unrecognized channel name is reported in the existing `rejected` array as `unknown_channel` rather than closing the connection. Closing took down every other channel the client held over one bad name, and a client cannot always know which channels its role allows before asking. Forbidden and profile-scoped channels were already handled this way. `required_action` in the hello frame is `"none"` for a declared connection and `"subscribe"` otherwise; the web type is widened to match. No wire field changes type or disappears, so this stays additive under the v1 rules. Part of #523 AI disclosure: tool Claude Code, model claude-opus-5, fully AI-generated, reviewed and verified by the author before submission. Co-Authored-By: Claude Opus 5 (1M context) * fix(events): address review findings on declared-channel subscriptions Three findings from automated review, all verified against the code before acting on them. **Start the reader before declared-channel snapshots (regression).** configureWebSocket installs an absolute read deadline that only pongs extend, and gorilla processes pongs solely inside ReadMessage (conn.go:950, reached only via advanceFrame). The declared path built its snapshots before starting the reader goroutine, so a snapshot slower than the deadline — a loaded jobs/sessions/scans/history query — would kill an otherwise healthy connection the instant reading began. The handshake path never had this problem because its snapshots run downstream of an active reader. Regression test stalls a tasks snapshot past the deadline; it fails with the previous ordering. **Advertise the feature through a capability endpoint.** Adding a client-visible subscription mode without one leaves a read-only client unable to tell, before connecting, whether ?channels= will be honored: an older server ignores it and closes the connection after the grace period, so the client must retain the very handshake this removes. GET /api/v1/events/capability reports both modes, the grace period the handler actually enforces, and the known channels, following the existing per-subsystem convention. **Deduplicate rejections, not just acceptances.** Asking twice for one forbidden channel produced two identical `rejected` entries. Pre-existing — the dedup check sat after the rejection branches — but cheap to correct in the function this PR extracted. Part of #523 AI disclosure: tool Claude Code, model claude-opus-5, fully AI-generated, reviewed and verified by the author before submission. Co-Authored-By: Claude Opus 5 (1M context) * docs(events): frame the capability endpoint as a staleness probe Clients are expected to run a current build rather than negotiate down to an old server, so the endpoint is not a branch-on-capability contract. Its value is letting a client distinguish "this server does not do that" from "the connection failed" — the two are indistinguishable from the socket alone, since an older server ignores ?channels= and then closes on the grace period — so it can tell the user the deployment is out of date instead of failing opaquely. Comment-only; no behavior or wire change. Co-Authored-By: Claude Opus 5 (1M context) * fix(events): bound the subscribed answer and reap unsubscribed connections Second review pass on the declared-channel path. Four issues, all at the edges of the new URL surface rather than in the design itself. Rejections amplified the request. Making an unknown channel non-fatal removed the brake that used to close the connection on the first bad name, and every refusal quotes the name it refuses — so a large ?channels= of distinct garbage produced a far larger `subscribed` frame, buffered server-side. Cap a selection at 32 distinct channels, report the overrun once instead of per name, truncate an echoed name at 64 bytes, and set a 64 KiB read limit on the socket so an oversize frame cannot be buffered whole before it is rejected. The grace period was disarmed by declaring, not by subscribing. Both `?channels=` with no names and a non-admin naming only an admin channel came up subscribed to nothing and were never reaped, each holding a hub subscriber, two goroutines, and an envelope channel that every published event fans into. Disarm on holding a subscription instead. Selection now resolves before the hello frame — it is pure, so nothing moves ahead of the reader — which lets required_action say "subscribe" when the connection really does still owe one. Repeating the parameter dropped channels silently. `?channels=a&channels=b` honored only the first and reported nothing rejected. Read every occurrence. The capability endpoint advertised `plugins`, which is host-to-plugin runtime dispatch and is granted to no role. An admin following the endpoint's stated purpose got `forbidden` while already being admin, and the hardcoded "Admin access required" made it a dead end rather than a soft failure. Split evt.ClientChannels out of AllChannels, advertise that, and word the refusal so it does not promise a remedy that does not exist. A test pins that an admin can subscribe to everything the endpoint names. Each guard was verified to bite by reverting it and watching the test fail. Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Opus 5 (1M context) --- internal/api/handlers/events_capability.go | 66 ++ .../api/handlers/events_capability_test.go | 80 +++ internal/api/handlers/events_ws.go | 300 ++++++-- .../events_ws_declared_channels_test.go | 638 ++++++++++++++++++ internal/api/router.go | 1 + internal/events/types.go | 18 + web/src/api/types.ts | 9 +- 7 files changed, 1046 insertions(+), 66 deletions(-) create mode 100644 internal/api/handlers/events_capability.go create mode 100644 internal/api/handlers/events_capability_test.go create mode 100644 internal/api/handlers/events_ws_declared_channels_test.go diff --git a/internal/api/handlers/events_capability.go b/internal/api/handlers/events_capability.go new file mode 100644 index 00000000..afdbaedd --- /dev/null +++ b/internal/api/handlers/events_capability.go @@ -0,0 +1,66 @@ +package handlers + +import ( + "encoding/json" + "net/http" + + evt "github.com/Silo-Server/silo-server/internal/events" +) + +// eventsCapabilityResponse describes how a client may subscribe to the events +// websocket. +// +// Clients are expected to run a current build rather than negotiate down to an +// old server, so this is not a branch-on-capability contract. It is how a +// client tells the difference between "this server does not do that" and "the +// connection failed", which is what lets it say the deployment is out of date +// instead of failing opaquely: a server predating declared channels ignores +// ?channels=, answers required_action:"subscribe", and closes the connection +// after the grace period, which is indistinguishable from a broken socket. +type eventsCapabilityResponse struct { + SchemaVersion int `json:"schema_version"` + // SubscribeFrame reports the handshake: connect, then send a subscribe + // frame. Always true; named so a future removal is detectable rather than + // silent. + SubscribeFrame bool `json:"subscribe_frame"` + // DeclaredChannels reports that ?channels= is honored on connect, that such + // a connection is exempt from the subscribe grace period, and that its + // hello frame carries required_action:"none". + DeclaredChannels bool `json:"declared_channels"` + // SubscribeGracePeriodSeconds is how long a connection holding no + // subscription may stay silent before it is closed. 0 would mean no + // deadline. + SubscribeGracePeriodSeconds int `json:"subscribe_grace_period_seconds"` + // MaxRequestedChannels is the most channels one selection may name, on the + // URL or in a subscribe frame. Names past it are answered with a single + // too_many_channels rejection rather than one per name. + MaxRequestedChannels int `json:"max_requested_channels"` + // Channels is every channel a client may ask for on this server, + // independent of role — not every channel the server has: the plugins + // channel is host-to-plugin runtime dispatch and is granted to no role, so + // naming it here would advertise a request that can only be refused. What + // the caller may actually subscribe to arrives as available_channels in the + // hello frame, which is role-filtered. + Channels []evt.EventChannel `json:"channels"` +} + +// HandleCapability reports the events websocket's subscription capabilities. +// +// Per the v1 rules, new functionality is feature-detected rather than inferred +// from a version. This follows the existing per-subsystem convention +// (/notifications/capability, /playback/capability, /downloads/capability). +// +// A client that finds declared_channels false is talking to a server older than +// its own expectations; the useful response is to tell the user to update the +// server, not to silently fall back. +func (h *EventsHandler) HandleCapability(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(eventsCapabilityResponse{ + SchemaVersion: 1, + SubscribeFrame: true, + DeclaredChannels: true, + SubscribeGracePeriodSeconds: int(subscribeGracePeriod.Seconds()), + MaxRequestedChannels: maxRequestedChannels, + Channels: evt.ClientChannels, + }) +} diff --git a/internal/api/handlers/events_capability_test.go b/internal/api/handlers/events_capability_test.go new file mode 100644 index 00000000..8582f960 --- /dev/null +++ b/internal/api/handlers/events_capability_test.go @@ -0,0 +1,80 @@ +package handlers + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + evt "github.com/Silo-Server/silo-server/internal/events" +) + +func TestEventsCapabilityReportsDeclaredChannelSupport(t *testing.T) { + handler := &EventsHandler{} + rec := httptest.NewRecorder() + handler.HandleCapability(rec, httptest.NewRequest(http.MethodGet, "/events/capability", nil)) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + if ct := rec.Header().Get("Content-Type"); ct != "application/json" { + t.Errorf("Content-Type = %q, want application/json", ct) + } + + var got eventsCapabilityResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decoding capability: %v (%s)", err, rec.Body.String()) + } + + if !got.DeclaredChannels { + t.Error("declared_channels = false; a client cannot detect ?channels= support") + } + if !got.SubscribeFrame { + t.Error("subscribe_frame = false; the handshake is still supported") + } + if got.SchemaVersion != 1 { + t.Errorf("schema_version = %d, want 1", got.SchemaVersion) + } + + // The advertised grace period must be the one the handler actually + // enforces, or a client will size its handshake timeout against fiction. + if want := int(subscribeGracePeriod.Seconds()); got.SubscribeGracePeriodSeconds != want { + t.Errorf("subscribe_grace_period_seconds = %d, want %d", got.SubscribeGracePeriodSeconds, want) + } + + if got.MaxRequestedChannels != maxRequestedChannels { + t.Errorf("max_requested_channels = %d, want %d", got.MaxRequestedChannels, maxRequestedChannels) + } + + if len(got.Channels) != len(evt.ClientChannels) { + t.Errorf("channels = %v, want all %d client channels", got.Channels, len(evt.ClientChannels)) + } +} + +// TestEventsCapabilityAdvertisesOnlySubscribableChannels is the point of +// publishing the list at all: a client that requests everything the endpoint +// names must not be refused any of it. The plugins channel is the case — it is +// in evt.AllChannels but granted to no role, admin included, so advertising it +// would send a client to a request that can only fail. +func TestEventsCapabilityAdvertisesOnlySubscribableChannels(t *testing.T) { + handler := &EventsHandler{} + rec := httptest.NewRecorder() + handler.HandleCapability(rec, httptest.NewRequest(http.MethodGet, "/events/capability", nil)) + + var got eventsCapabilityResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decoding capability: %v (%s)", err, rec.Body.String()) + } + + // An admin, on a profile-bound connection, is the most permissive caller + // there is. Every advertised channel must resolve for them. + _, accepted, rejected := resolveChannelSelection( + got.Channels, allowedChannelsForRole("admin"), "profile-1") + + if len(rejected) != 0 { + t.Errorf("capability advertises channels an admin cannot subscribe to: %v", rejected) + } + if len(accepted) != len(got.Channels) { + t.Errorf("accepted %d of %d advertised channels", len(accepted), len(got.Channels)) + } +} diff --git a/internal/api/handlers/events_ws.go b/internal/api/handlers/events_ws.go index 52426179..b9c8d09c 100644 --- a/internal/api/handlers/events_ws.go +++ b/internal/api/handlers/events_ws.go @@ -5,6 +5,9 @@ import ( "encoding/json" "log/slog" "net/http" + "net/url" + "strconv" + "strings" "time" "github.com/Silo-Server/silo-server/internal/adminjob" @@ -30,6 +33,41 @@ type taskInfoLister interface { const maxRealtimeScanSnapshotRuns = 500 +// subscribeGracePeriod bounds how long a connection that is not subscribed to +// anything may stay silent before it is closed. It exists to distinguish a +// client mid-handshake from one that connected and stalled; a connection that +// arrived with a usable ?channels= selection is never put on this clock. +const subscribeGracePeriod = 5 * time.Second + +// maxRequestedChannels bounds how many distinct channels one selection may +// name. Every refusal is echoed back with its channel name, so without a cap a +// selection of many unknown names is an amplifier: the request is bounded by +// the header/frame limit, the answer was not. Any real client names a subset of +// the ten client channels; the slack is only so a caller learns it overshot +// rather than silently losing channels. +const maxRequestedChannels = 32 + +// maxChannelNameLength bounds the channel name echoed in a rejection. A name +// longer than any real channel is garbage by definition, and quoting it back in +// full is the second half of the same amplification. +const maxChannelNameLength = 64 + +// maxEventsFrameBytes bounds an inbound frame on this socket. The only message +// it accepts is a subscribe frame naming at most maxRequestedChannels channels, +// which is orders of magnitude smaller; the limit exists so a client cannot +// make the server buffer an arbitrarily large frame before it is rejected. +const maxEventsFrameBytes = 64 * 1024 + +// required_action values in the hello frame: whether the client still owes a +// subscribe frame, or already holds a subscription it declared on the URL. +const ( + requiredActionSubscribe = "subscribe" + requiredActionNone = "none" +) + +// frameTypeSubscribed is the acknowledgement both selection paths answer with. +const frameTypeSubscribed = "subscribed" + type activeScanLister interface { ListActiveSnapshot(ctx context.Context, limit int) ([]evt.ScanRun, error) } @@ -127,6 +165,7 @@ func (h *EventsHandler) HandleWebSocket(w http.ResponseWriter, r *http.Request) } defer conn.Close() configureWebSocket(conn) + conn.SetReadLimit(maxEventsFrameBytes) ctx, cancel := context.WithCancel(r.Context()) defer cancel() @@ -138,17 +177,52 @@ func (h *EventsHandler) HandleWebSocket(w http.ResponseWriter, r *http.Request) }) allowedChannels := allowedChannelsForRole(claims.Role) + + // A connection may declare its channels on the URL instead of sending a + // subscribe frame. That is all an observer needs — subscribe is the only + // inbound message this endpoint accepts — so declaring up front lets a + // read-only client skip the handshake, the grace period, and the write + // half of the socket entirely. + declaredChannels, declared := parseDeclaredChannels(r.URL.Query()) + + // Resolved before the hello frame so required_action can report what the + // connection actually owes. Resolution is pure — no query, no I/O — so + // nothing here can stall ahead of the reader that starts below; only the + // snapshots the accepted channels produce can, and those still run after + // it. + declaredSubs := make(map[evt.EventChannel]struct{}) + var declaredAccepted []evt.EventChannel + var declaredRejected []evt.EventsRejectedChannel + if declared { + declaredSubs, declaredAccepted, declaredRejected = + resolveChannelSelection(declaredChannels, allowedChannels, boundProfileID) + } + + // A declaration that yielded no subscription discharges nothing: the client + // still owes a subscribe frame, and saying "none" would send it to wait + // silently for events on a connection due to be closed in five seconds. + requiredAction := requiredActionSubscribe + if len(declaredSubs) > 0 { + requiredAction = requiredActionNone + } connectionID := ulid.Make().String() if err := writeWebSocketJSON(conn, evt.EventsHelloMessage{ Type: "hello", SchemaVersion: 1, ConnectionID: connectionID, AvailableChannels: allowedChannels, - RequiredAction: "subscribe", + RequiredAction: requiredAction, }); err != nil { return } + // The reader starts before any snapshot query runs. configureWebSocket + // installs an absolute read deadline that only pongs extend, and gorilla + // processes pongs solely inside ReadMessage — so a snapshot slow enough to + // outlast the deadline (a loaded jobs/sessions/scans query) would kill an + // otherwise healthy connection the moment reading began. The handshake path + // is safe for free, building its snapshots downstream of an active reader; + // the declared path has to arrange that ordering deliberately. readMessages := make(chan []byte, 8) readDone := make(chan struct{}) go func() { @@ -166,11 +240,40 @@ func (h *EventsHandler) HandleWebSocket(w http.ResponseWriter, r *http.Request) } }() - deadline := time.NewTimer(5 * time.Second) - defer deadline.Stop() + subscriptions := declaredSubs + if declared { + // The same subscribed frame the handshake produces, so a client sees + // one acknowledgement shape however it selected its channels — and + // still learns which of its requests were refused, and why. + if err := writeWebSocketJSON(conn, evt.EventsSubscribedMessage{ + Type: frameTypeSubscribed, + Channels: declaredAccepted, + Rejected: declaredRejected, + }); err != nil { + return + } + for _, channel := range declaredAccepted { + if err := h.writeSnapshotFrame(conn, r, claims, boundProfileID, channel); err != nil { + return + } + } + } - subscriptions := make(map[evt.EventChannel]struct{}) - subscribedOnce := false + // The clock is disarmed by holding a subscription, not by having declared + // one. A declaration that resolved to nothing — ?channels= with no names, + // or a selection every entry of which was refused — leaves the connection + // in exactly the state the grace period exists to reap: subscribed to + // nothing, with no reason to expect it to ever receive a frame, while still + // holding a hub subscriber, two goroutines, and an envelope channel that + // every published event fans into. Such a client still owes a subscribe + // frame, and its rejected list told it why. A nil channel blocks forever, + // which is exactly the disarmed case. + var deadlineC <-chan time.Time + if len(subscriptions) == 0 { + deadline := time.NewTimer(subscribeGracePeriod) + defer deadline.Stop() + deadlineC = deadline.C + } for { select { @@ -178,11 +281,8 @@ func (h *EventsHandler) HandleWebSocket(w http.ResponseWriter, r *http.Request) return case <-readDone: return - case <-deadline.C: - if subscribedOnce { - continue - } - writeWebSocketError(conn, "bad_request", "subscribe is required within 5 seconds") + case <-deadlineC: + writeWebSocketError(conn, "bad_request", "subscribe is required within "+subscribeGracePeriod.String()) _ = writeWebSocketControl( conn, websocket.CloseMessage, @@ -197,13 +297,10 @@ func (h *EventsHandler) HandleWebSocket(w http.ResponseWriter, r *http.Request) if !handled { continue } - subscribedOnce = true - if !deadline.Stop() { - select { - case <-deadline.C: - default: - } - } + // Obligation discharged: detach the timer so a later tick cannot + // close a connection that has since subscribed. The timer itself + // is stopped by the deferred Stop above. + deadlineC = nil subscriptions = nextSubs case env, ok := <-eventsCh: if !ok { @@ -263,56 +360,10 @@ func (h *EventsHandler) handleEventsClientMessage( return nil, false, false } - allowedSet := make(map[evt.EventChannel]struct{}, len(allowed)) - for _, channel := range allowed { - allowedSet[channel] = struct{}{} - } - validSet := make(map[evt.EventChannel]struct{}, len(evt.AllChannels)) - for _, channel := range evt.AllChannels { - validSet[channel] = struct{}{} - } - - nextSubs := make(map[evt.EventChannel]struct{}, len(message.Channels)) - accepted := make([]evt.EventChannel, 0, len(message.Channels)) - rejected := make([]evt.EventsRejectedChannel, 0) - - for _, channel := range message.Channels { - if _, ok := validSet[channel]; !ok { - writeWebSocketError(conn, "bad_request", "Invalid channel") - _ = writeWebSocketControl( - conn, - websocket.CloseMessage, - websocket.FormatCloseMessage(websocket.ClosePolicyViolation, "invalid channel"), - ) - return nil, false, false - } - if _, ok := allowedSet[channel]; !ok { - rejected = append(rejected, evt.EventsRejectedChannel{ - Channel: channel, - Code: "forbidden", - Message: "Admin access required", - }) - continue - } - // The notifications channel is profile-scoped: it requires a - // connection bound to a profile via a websocket ticket. - if channel == evt.ChannelNotifications && boundProfileID == "" { - rejected = append(rejected, evt.EventsRejectedChannel{ - Channel: channel, - Code: "profile_required", - Message: "A profile-bound websocket ticket is required", - }) - continue - } - if _, seen := nextSubs[channel]; seen { - continue - } - nextSubs[channel] = struct{}{} - accepted = append(accepted, channel) - } + nextSubs, accepted, rejected := resolveChannelSelection(message.Channels, allowed, boundProfileID) if err := writeWebSocketJSON(conn, evt.EventsSubscribedMessage{ - Type: "subscribed", + Type: frameTypeSubscribed, RequestID: message.RequestID, Channels: accepted, Rejected: rejected, @@ -329,6 +380,125 @@ func (h *EventsHandler) handleEventsClientMessage( return nextSubs, true, true } +// resolveChannelSelection decides which of the requested channels a connection +// may subscribe to. It is the single place that answers that question, shared +// by the URL-declared path and the subscribe frame so the two cannot drift. +// +// Every rejection is reported rather than fatal: an unknown, forbidden, or +// profile-requiring channel costs the caller that channel and nothing else. +// Closing the connection instead would take down every other channel it holds +// over one bad name, and a client cannot always know which channels its role +// allows before it asks. +// +// Because refusals are answered instead of closing the connection, the answer +// has to be bounded independently of the request: at most maxRequestedChannels +// distinct names are considered, and an overrun is reported once rather than +// per excess name. +func resolveChannelSelection( + requested []evt.EventChannel, + allowed []evt.EventChannel, + boundProfileID string, +) (map[evt.EventChannel]struct{}, []evt.EventChannel, []evt.EventsRejectedChannel) { + allowedSet := make(map[evt.EventChannel]struct{}, len(allowed)) + for _, channel := range allowed { + allowedSet[channel] = struct{}{} + } + validSet := make(map[evt.EventChannel]struct{}, len(evt.ClientChannels)) + for _, channel := range evt.ClientChannels { + validSet[channel] = struct{}{} + } + + capacity := min(len(requested), maxRequestedChannels) + subs := make(map[evt.EventChannel]struct{}, capacity) + accepted := make([]evt.EventChannel, 0, capacity) + rejected := make([]evt.EventsRejectedChannel, 0, capacity) + // A repeated channel is answered once whether it was accepted or refused. + // Only the accepted side deduplicated before this change, so asking twice + // for one forbidden channel produced two identical rejections. + seen := make(map[evt.EventChannel]struct{}, capacity) + + reject := func(channel evt.EventChannel, code, message string) { + // A garbage name is echoed back only far enough to be recognizable. + if len(channel) > maxChannelNameLength { + channel = channel[:maxChannelNameLength] + } + rejected = append(rejected, evt.EventsRejectedChannel{ + Channel: channel, + Code: code, + Message: message, + }) + } + + for _, channel := range requested { + if _, dup := seen[channel]; dup { + continue + } + if len(seen) >= maxRequestedChannels { + // One entry for the whole overrun, not one per name: the point of + // the cap is that the response cannot grow with the request. + reject("", "too_many_channels", + "At most "+strconv.Itoa(maxRequestedChannels)+" channels may be requested at once") + break + } + seen[channel] = struct{}{} + + switch { + case !contains(validSet, channel): + reject(channel, "unknown_channel", "Unknown channel") + case !contains(allowedSet, channel): + // Deliberately not "admin access required": the caller's role is + // the usual reason a channel is refused but not the only one, and a + // remedy that does not exist reads as a bug in the client's own + // permissions rather than a channel it was never going to get. + reject(channel, "forbidden", "This channel is not available to this connection") + // The notifications channel is profile-scoped: it requires a + // connection bound to a profile via a websocket ticket. + case channel == evt.ChannelNotifications && boundProfileID == "": + reject(channel, "profile_required", "A profile-bound websocket ticket is required") + default: + subs[channel] = struct{}{} + accepted = append(accepted, channel) + } + } + + return subs, accepted, rejected +} + +func contains(set map[evt.EventChannel]struct{}, channel evt.EventChannel) bool { + _, ok := set[channel] + return ok +} + +// parseDeclaredChannels reads the optional ?channels= selection a connection +// may declare on the URL, letting a read-only observer skip the subscribe +// handshake entirely. An absent parameter returns ok=false, which keeps the +// handshake (and its grace period) in force; an empty or all-blank value is a +// deliberate declaration of nothing, so it returns ok=true and no channels. +// +// Every occurrence of the parameter is read, not just the first. Repeating a +// query parameter is as natural a spelling as a comma-separated list, and +// honoring one and silently discarding the rest loses channels with no +// diagnostic — the connection would come up subscribed to less than it asked +// for and report nothing rejected. +func parseDeclaredChannels(query url.Values) ([]evt.EventChannel, bool) { + values, ok := query["channels"] + if !ok { + return nil, false + } + + channels := make([]evt.EventChannel, 0, 4) + for _, value := range values { + for _, raw := range strings.Split(value, ",") { + name := strings.TrimSpace(raw) + if name == "" { + continue + } + channels = append(channels, evt.EventChannel(name)) + } + } + return channels, true +} + func allowedChannelsForRole(role string) []evt.EventChannel { channels := []evt.EventChannel{ evt.ChannelCatalog, diff --git a/internal/api/handlers/events_ws_declared_channels_test.go b/internal/api/handlers/events_ws_declared_channels_test.go new file mode 100644 index 00000000..a2217cb5 --- /dev/null +++ b/internal/api/handlers/events_ws_declared_channels_test.go @@ -0,0 +1,638 @@ +package handlers + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + + 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/cache" + evt "github.com/Silo-Server/silo-server/internal/events" + "github.com/Silo-Server/silo-server/internal/taskmanager" +) + +// eventsWSTestConn dials the events websocket against a handler authenticated +// as the given claims, and returns a frame reader. These tests go through the +// real socket rather than calling the handler directly because the behavior +// under test — what a connection is subscribed to before it has said anything, +// and whether it survives the grace period — only exists in the connection +// loop. +func eventsWSTestConn(t *testing.T, hub *evt.Hub, claims *auth.Claims, query string) ( + *websocket.Conn, + func(wantType string) map[string]json.RawMessage, +) { + t.Helper() + return eventsWSTestConnWithHandler(t, &EventsHandler{hub: hub}, claims, query) +} + +func eventsWSTestConnWithHandler( + t *testing.T, + handler *EventsHandler, + claims *auth.Claims, + query string, +) (*websocket.Conn, func(wantType string) map[string]json.RawMessage) { + t.Helper() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ctx := apimw.SetClaims(r.Context(), claims) + handler.HandleWebSocket(w, r.WithContext(ctx)) + })) + t.Cleanup(server.Close) + + conn, resp, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http")+query, nil) + if err != nil { + t.Fatalf("dialing events websocket: %v", err) + } + if resp != nil && resp.Body != nil { + t.Cleanup(func() { _ = resp.Body.Close() }) + } + t.Cleanup(func() { _ = conn.Close() }) + + readFrame := func(wantType string) map[string]json.RawMessage { + t.Helper() + if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatalf("setting read deadline: %v", err) + } + _, data, err := conn.ReadMessage() + if err != nil { + t.Fatalf("reading %s frame: %v", wantType, err) + } + var frame map[string]json.RawMessage + if err := json.Unmarshal(data, &frame); err != nil { + t.Fatalf("frame is not JSON: %v (%s)", err, data) + } + if string(frame["type"]) != `"`+wantType+`"` { + t.Fatalf("frame type = %s, want %q (frame: %s)", frame["type"], wantType, data) + } + return frame + } + + return conn, readFrame +} + +// TestEventsWebSocketDeclaredChannelsSkipHandshake is the point of the feature: +// a connection that named its channels on the URL receives events without ever +// writing to the socket. +func TestEventsWebSocketDeclaredChannelsSkipHandshake(t *testing.T) { + hub := evt.NewHub("test", &cache.NoopEventBus{}) + _, readFrame := eventsWSTestConn(t, hub, + &auth.Claims{UserID: 1, Role: "user"}, "?channels=user_settings") + + hello := readFrame("hello") + if string(hello["required_action"]) != `"none"` { + t.Errorf("required_action = %s, want \"none\"", hello["required_action"]) + } + + subscribed := readFrame("subscribed") + if !strings.Contains(string(subscribed["channels"]), `"user_settings"`) { + t.Fatalf("declared channel was not accepted: %s", subscribed["channels"]) + } + + // Accepted channels hydrate with a snapshot, exactly as the handshake does. + if snapshot := readFrame("snapshot"); string(snapshot["channel"]) != `"user_settings"` { + t.Fatalf("snapshot channel = %s, want user_settings", snapshot["channel"]) + } + + publishUserSettingsEvent(context.Background(), hub, 1, "profile-1", + "playback.subtitle_language", "profile") + + if event := readFrame("event"); string(event["channel"]) != `"user_settings"` { + t.Errorf("event channel = %s, want user_settings", event["channel"]) + } +} + +// TestEventsWebSocketDeclaredChannelsSurviveGracePeriod guards the core promise +// of the change: an observer that connects and never speaks stays connected. +// Before this, it was closed with a policy violation after five seconds. +func TestEventsWebSocketDeclaredChannelsSurviveGracePeriod(t *testing.T) { + if testing.Short() { + t.Skip("waits out the subscribe grace period in real time") + } + + hub := evt.NewHub("test", &cache.NoopEventBus{}) + conn, readFrame := eventsWSTestConn(t, hub, + &auth.Claims{UserID: 1, Role: "user"}, "?channels=user_settings") + + readFrame("hello") + readFrame("subscribed") + readFrame("snapshot") + + // Stay silent well past the deadline that would have closed a + // handshake-style connection, then confirm the socket still delivers. + time.Sleep(subscribeGracePeriod + time.Second) + + publishUserSettingsEvent(context.Background(), hub, 1, "profile-1", + "playback.subtitle_language", "profile") + + if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatalf("setting read deadline: %v", err) + } + _, data, err := conn.ReadMessage() + if err != nil { + t.Fatalf("connection did not survive the grace period: %v", err) + } + var frame map[string]json.RawMessage + if err := json.Unmarshal(data, &frame); err != nil { + t.Fatalf("frame is not JSON: %v (%s)", err, data) + } + if string(frame["type"]) != `"event"` { + t.Fatalf("frame type = %s, want \"event\" (frame: %s)", frame["type"], data) + } +} + +// TestEventsWebSocketEmptyDeclarationStillClosed pins what disarms the grace +// period: holding a subscription, not having spelled ?channels=. A declaration +// that resolved to nothing leaves the connection in the exact state the clock +// exists to reap — no subscriptions, no reason to expect a frame, but still a +// hub subscriber, two goroutines, and an envelope channel every published +// event fans into. +func TestEventsWebSocketEmptyDeclarationStillClosed(t *testing.T) { + if testing.Short() { + t.Skip("waits out the subscribe grace period in real time") + } + + tests := []struct { + name string + query string + }{ + {name: "declares no channels", query: "?channels="}, + // Every name refused: a non-admin naming only an admin channel is + // subscribed to nothing, exactly as if it had named nothing. + {name: "every declared channel refused", query: "?channels=sessions"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + hub := evt.NewHub("test", &cache.NoopEventBus{}) + conn, readFrame := eventsWSTestConn(t, hub, + &auth.Claims{UserID: 1, Role: "user"}, tt.query) + + hello := readFrame("hello") + // The obligation is real, so the hello frame has to say so rather + // than sending the client off to wait silently on a doomed socket. + if string(hello["required_action"]) != `"subscribe"` { + t.Errorf("required_action = %s, want \"subscribe\"", hello["required_action"]) + } + readFrame("subscribed") + + if err := conn.SetReadDeadline(time.Now().Add(subscribeGracePeriod + 5*time.Second)); err != nil { + t.Fatalf("setting read deadline: %v", err) + } + // The error frame, then the close. + if _, _, err := conn.ReadMessage(); err != nil { + t.Fatalf("reading error frame: %v", err) + } + if _, _, err := conn.ReadMessage(); !websocket.IsCloseError(err, websocket.ClosePolicyViolation) { + t.Fatalf("close error = %v, want policy violation", err) + } + }) + } +} + +// TestEventsWebSocketPartialDeclarationSurvives is the boundary case on the +// other side: one accepted channel among refusals is a live subscription, so +// the connection is not on the clock. +func TestEventsWebSocketPartialDeclarationSurvives(t *testing.T) { + if testing.Short() { + t.Skip("waits out the subscribe grace period in real time") + } + + hub := evt.NewHub("test", &cache.NoopEventBus{}) + conn, readFrame := eventsWSTestConn(t, hub, + &auth.Claims{UserID: 1, Role: "user"}, "?channels=sessions,user_settings") + + hello := readFrame("hello") + if string(hello["required_action"]) != `"none"` { + t.Errorf("required_action = %s, want \"none\"", hello["required_action"]) + } + readFrame("subscribed") + readFrame("snapshot") + + time.Sleep(subscribeGracePeriod + time.Second) + + publishUserSettingsEvent(context.Background(), hub, 1, "profile-1", + "playback.subtitle_language", "profile") + + if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatalf("setting read deadline: %v", err) + } + if _, _, err := conn.ReadMessage(); err != nil { + t.Fatalf("a partially accepted declaration did not survive the grace period: %v", err) + } +} + +// TestEventsWebSocketRepeatedChannelsParameter covers the other natural +// spelling of a selection. Honoring only the first occurrence dropped the rest +// with an empty rejected array — the connection came up subscribed to less than +// it asked for and reported nothing wrong. +func TestEventsWebSocketRepeatedChannelsParameter(t *testing.T) { + hub := evt.NewHub("test", &cache.NoopEventBus{}) + _, readFrame := eventsWSTestConn(t, hub, &auth.Claims{UserID: 1, Role: "user"}, + "?channels=catalog&channels=user_settings") + + readFrame("hello") + subscribed := readFrame("subscribed") + + for _, want := range []string{`"catalog"`, `"user_settings"`} { + if !strings.Contains(string(subscribed["channels"]), want) { + t.Errorf("channel %s was dropped: %s", want, subscribed["channels"]) + } + } +} + +// TestEventsWebSocketSilentConnectionStillClosed pins the other half: the grace +// period still applies to a connection that declared nothing, so the URL path +// relaxes the rule rather than removing it. +func TestEventsWebSocketSilentConnectionStillClosed(t *testing.T) { + if testing.Short() { + t.Skip("waits out the subscribe grace period in real time") + } + + hub := evt.NewHub("test", &cache.NoopEventBus{}) + conn, readFrame := eventsWSTestConn(t, hub, &auth.Claims{UserID: 1, Role: "user"}, "") + + hello := readFrame("hello") + if string(hello["required_action"]) != `"subscribe"` { + t.Errorf("required_action = %s, want \"subscribe\"", hello["required_action"]) + } + + if err := conn.SetReadDeadline(time.Now().Add(subscribeGracePeriod + 5*time.Second)); err != nil { + t.Fatalf("setting read deadline: %v", err) + } + // The error frame, then the close. + if _, _, err := conn.ReadMessage(); err != nil { + t.Fatalf("reading error frame: %v", err) + } + _, _, err := conn.ReadMessage() + if err == nil { + t.Fatal("silent connection was not closed after the grace period") + } + if !websocket.IsCloseError(err, websocket.ClosePolicyViolation) { + t.Fatalf("close error = %v, want policy violation", err) + } +} + +// TestEventsWebSocketDeclaredChannelsCannotEscalate is the authorization +// guarantee: naming an admin-only channel on the URL must not grant it, and +// must not grant the events published to it either. +func TestEventsWebSocketDeclaredChannelsCannotEscalate(t *testing.T) { + hub := evt.NewHub("test", &cache.NoopEventBus{}) + _, readFrame := eventsWSTestConn(t, hub, + &auth.Claims{UserID: 1, Role: "user"}, "?channels=sessions,user_settings") + + readFrame("hello") + subscribed := readFrame("subscribed") + + if strings.Contains(string(subscribed["channels"]), `"sessions"`) { + t.Fatalf("non-admin was granted the sessions channel: %s", subscribed["channels"]) + } + if !strings.Contains(string(subscribed["rejected"]), `"forbidden"`) { + t.Errorf("sessions was not reported as forbidden: %s", subscribed["rejected"]) + } + // The permitted channel in the same request still landed. + if !strings.Contains(string(subscribed["channels"]), `"user_settings"`) { + t.Errorf("a forbidden channel denied the rest of the request: %s", subscribed["channels"]) + } + + readFrame("snapshot") // user_settings + + // An admin-only event on the refused channel must not be delivered. Publish + // it first, then a permitted event; receiving the second without the first + // proves the first was filtered rather than merely slow. + if err := hub.PublishJSON(context.Background(), evt.ChannelSessions, "sessions.replaced", nil, + evt.PublishOptions{AdminOnly: true}); err != nil { + t.Fatalf("publishing admin-only event: %v", err) + } + publishUserSettingsEvent(context.Background(), hub, 1, "profile-1", + "playback.subtitle_language", "profile") + + event := readFrame("event") + if string(event["channel"]) != `"user_settings"` { + t.Fatalf("received an event on a channel this connection was refused: %s", event["channel"]) + } +} + +// TestEventsWebSocketUnknownChannelDoesNotCloseConnection covers the third +// change: an unrecognized channel name used to close the socket outright, +// taking down every other channel the client held over one bad name. +func TestEventsWebSocketUnknownChannelDoesNotCloseConnection(t *testing.T) { + hub := evt.NewHub("test", &cache.NoopEventBus{}) + conn, readFrame := eventsWSTestConn(t, hub, &auth.Claims{UserID: 1, Role: "user"}, "") + + readFrame("hello") + + if err := conn.WriteJSON(evt.EventsSubscribeMessage{ + Type: "subscribe", + RequestID: "r1", + Channels: []evt.EventChannel{"not_a_channel", evt.ChannelUserSettings}, + }); err != nil { + t.Fatalf("sending subscribe: %v", err) + } + + subscribed := readFrame("subscribed") + if !strings.Contains(string(subscribed["rejected"]), `"unknown_channel"`) { + t.Errorf("unknown channel was not reported as such: %s", subscribed["rejected"]) + } + if !strings.Contains(string(subscribed["channels"]), `"user_settings"`) { + t.Fatalf("an unknown channel denied the valid one alongside it: %s", subscribed["channels"]) + } + + // The connection is still usable. + readFrame("snapshot") + publishUserSettingsEvent(context.Background(), hub, 1, "profile-1", + "playback.subtitle_language", "profile") + if event := readFrame("event"); string(event["channel"]) != `"user_settings"` { + t.Errorf("event channel = %s, want user_settings", event["channel"]) + } +} + +// TestEventsWebSocketRejectsOversizeFrame covers the one case that is still +// fatal, and has to be: a frame is buffered whole before its type can be read, +// so an oversize frame cannot be answered with a rejection the way a bad +// channel name can — refusing it politely would mean first doing the thing the +// limit exists to prevent. +func TestEventsWebSocketRejectsOversizeFrame(t *testing.T) { + hub := evt.NewHub("test", &cache.NoopEventBus{}) + conn, readFrame := eventsWSTestConn(t, hub, &auth.Claims{UserID: 1, Role: "user"}, "") + readFrame("hello") + + oversize := `{"type":"subscribe","channels":["` + + strings.Repeat("x", maxEventsFrameBytes*2) + `"]}` + // The write itself may fail once the server has already torn the connection + // down, which is the same outcome; only accepting the frame is a failure. + if err := conn.WriteMessage(websocket.TextMessage, []byte(oversize)); err != nil { + return + } + if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatalf("setting read deadline: %v", err) + } + if _, _, err := conn.ReadMessage(); err == nil { + t.Fatal("a frame past the read limit was accepted") + } +} + +// slowTaskLister stalls the tasks snapshot long enough to outlast the read +// deadline configureWebSocket installs at connect. +type slowTaskLister struct{ delay time.Duration } + +func (s slowTaskLister) ListTasks(bool) []taskmanager.TaskInfo { + time.Sleep(s.delay) + return []taskmanager.TaskInfo{} +} + +// TestEventsWebSocketDeclaredChannelsSurviveSlowSnapshot pins the ordering the +// declared path depends on. configureWebSocket sets an absolute read deadline +// that only pongs extend, and gorilla processes pongs solely inside +// ReadMessage — so if snapshot queries ran before the reader goroutine started, +// a snapshot slower than the deadline would kill a healthy connection the +// instant reading began. The handshake path gets this for free by building +// snapshots downstream of an active reader; the declared path arranges it +// deliberately. +func TestEventsWebSocketDeclaredChannelsSurviveSlowSnapshot(t *testing.T) { + if testing.Short() { + t.Skip("stalls a snapshot past the websocket read deadline in real time") + } + + hub := evt.NewHub("test", &cache.NoopEventBus{}) + handler := &EventsHandler{ + hub: hub, + tasks: slowTaskLister{delay: wsPingInterval + wsPongTimeout + 2*time.Second}, + } + + conn, readFrame := eventsWSTestConnWithHandler(t, handler, + &auth.Claims{UserID: 1, Role: "admin"}, "?channels=tasks") + + readFrame("hello") + readFrame("subscribed") + + // The snapshot arrives late by design; allow for the stall plus slack. + if err := conn.SetReadDeadline(time.Now().Add(wsPingInterval + wsPongTimeout + 15*time.Second)); err != nil { + t.Fatalf("setting read deadline: %v", err) + } + _, data, err := conn.ReadMessage() + if err != nil { + t.Fatalf("connection died during a slow snapshot: %v", err) + } + var frame map[string]json.RawMessage + if err := json.Unmarshal(data, &frame); err != nil { + t.Fatalf("frame is not JSON: %v (%s)", err, data) + } + if string(frame["type"]) != `"snapshot"` { + t.Fatalf("frame type = %s, want \"snapshot\" (frame: %s)", frame["type"], data) + } + + // And the connection is still live afterwards. + if err := hub.PublishJSON(context.Background(), evt.ChannelTasks, "tasks.changed", + map[string]string{"id": "t1"}, evt.PublishOptions{}); err != nil { + t.Fatalf("publishing: %v", err) + } + if event := readFrame("event"); string(event["channel"]) != `"tasks"` { + t.Errorf("event channel = %s, want tasks", event["channel"]) + } +} + +func TestParseDeclaredChannels(t *testing.T) { + tests := []struct { + name string + query string + want []evt.EventChannel + declared bool + }{ + { + name: "absent parameter keeps the handshake", + query: "", + want: nil, + declared: false, + }, + { + name: "empty value declares nothing, but still declares", + query: "channels=", + want: []evt.EventChannel{}, + declared: true, + }, + { + name: "whitespace and empty entries are dropped", + query: "channels=catalog,%20,,user_state%20", + want: []evt.EventChannel{evt.ChannelCatalog, evt.ChannelUserState}, + declared: true, + }, + { + // Repeating the parameter is as natural a spelling as one comma + // list; reading only the first occurrence lost the rest silently. + name: "every occurrence of the parameter is read", + query: "channels=catalog&channels=user_state,user_settings", + want: []evt.EventChannel{evt.ChannelCatalog, evt.ChannelUserState, evt.ChannelUserSettings}, + declared: true, + }, + { + name: "a repeated parameter with only blank values still declares", + query: "channels=&channels=%20", + want: []evt.EventChannel{}, + declared: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + query, err := url.ParseQuery(tt.query) + if err != nil { + t.Fatalf("parsing query: %v", err) + } + got, declared := parseDeclaredChannels(query) + if declared != tt.declared { + t.Fatalf("declared = %v, want %v", declared, tt.declared) + } + if len(got) != len(tt.want) { + t.Fatalf("channels = %v, want %v", got, tt.want) + } + for i := range got { + if got[i] != tt.want[i] { + t.Fatalf("channels = %v, want %v", got, tt.want) + } + } + }) + } +} + +func TestResolveChannelSelectionDeniesByDefault(t *testing.T) { + allowed := allowedChannelsForRole("user") + + subs, accepted, rejected := resolveChannelSelection( + []evt.EventChannel{evt.ChannelSessions, "bogus", evt.ChannelNotifications}, + allowed, + "", // unbound: notifications requires a profile-bound ticket + ) + + if len(subs) != 0 || len(accepted) != 0 { + t.Fatalf("nothing should have been accepted: subs=%v accepted=%v", subs, accepted) + } + codes := make(map[string]evt.EventChannel, len(rejected)) + for _, r := range rejected { + codes[r.Code] = r.Channel + } + if codes["forbidden"] != evt.ChannelSessions { + t.Errorf("sessions not rejected as forbidden: %v", rejected) + } + if codes["unknown_channel"] != "bogus" { + t.Errorf("bogus not rejected as unknown: %v", rejected) + } + if codes["profile_required"] != evt.ChannelNotifications { + t.Errorf("notifications not rejected as profile_required: %v", rejected) + } +} + +func TestResolveChannelSelectionDeduplicates(t *testing.T) { + subs, accepted, rejected := resolveChannelSelection( + []evt.EventChannel{evt.ChannelJobs, evt.ChannelJobs}, + allowedChannelsForRole("admin"), + "", + ) + + if len(subs) != 1 || len(accepted) != 1 { + t.Fatalf("duplicate channel was not collapsed: subs=%v accepted=%v", subs, accepted) + } + if len(rejected) != 0 { + t.Fatalf("unexpected rejections: %v", rejected) + } +} + +// TestResolveChannelSelectionDeduplicatesRejections covers the other half of +// dedup: a channel asked for twice is answered once whether it was accepted or +// refused. Only the accepted side deduplicated before. +func TestResolveChannelSelectionDeduplicatesRejections(t *testing.T) { + _, _, rejected := resolveChannelSelection( + []evt.EventChannel{ + evt.ChannelSessions, evt.ChannelSessions, // forbidden for a user + "bogus", "bogus", // unknown + }, + allowedChannelsForRole("user"), + "", + ) + + if len(rejected) != 2 { + t.Fatalf("rejected = %v, want one entry per distinct channel", rejected) + } +} + +// TestResolveChannelSelectionBoundsTheAnswer is the amplification guard. +// Refusals quote the name they refuse, so once an unknown channel stopped +// closing the connection, the response grew with the request: a large selection +// of distinct garbage names produced a far larger subscribed frame, buffered +// server-side. The answer has to be bounded independently of the request. +func TestResolveChannelSelectionBoundsTheAnswer(t *testing.T) { + requested := make([]evt.EventChannel, 0, 5000) + for i := range 5000 { + requested = append(requested, evt.EventChannel("bogus-"+strconv.Itoa(i))) + } + + subs, accepted, rejected := resolveChannelSelection(requested, allowedChannelsForRole("user"), "") + + if len(subs) != 0 || len(accepted) != 0 { + t.Fatalf("garbage names were accepted: subs=%v accepted=%v", subs, accepted) + } + // Every considered name is refused, plus exactly one entry for the overrun. + if len(rejected) != maxRequestedChannels+1 { + t.Fatalf("rejected %d entries, want %d", len(rejected), maxRequestedChannels+1) + } + overrun := rejected[len(rejected)-1] + if overrun.Code != "too_many_channels" { + t.Errorf("last rejection code = %q, want too_many_channels", overrun.Code) + } + + // The response must not scale with the request, whatever the constants are. + encoded, err := json.Marshal(evt.EventsSubscribedMessage{ + Type: "subscribed", + Channels: accepted, + Rejected: rejected, + }) + if err != nil { + t.Fatalf("encoding subscribed frame: %v", err) + } + if len(encoded) > 8*1024 { + t.Errorf("subscribed frame is %d bytes for a garbage selection", len(encoded)) + } +} + +// TestResolveChannelSelectionTruncatesLongNames covers the per-name half of the +// same concern: one enormous name is as good an amplifier as many small ones. +func TestResolveChannelSelectionTruncatesLongNames(t *testing.T) { + long := evt.EventChannel(strings.Repeat("x", 4096)) + + _, _, rejected := resolveChannelSelection( + []evt.EventChannel{long}, allowedChannelsForRole("user"), "") + + if len(rejected) != 1 { + t.Fatalf("rejected = %v, want one entry", rejected) + } + if len(rejected[0].Channel) != maxChannelNameLength { + t.Errorf("echoed name is %d bytes, want it truncated to %d", + len(rejected[0].Channel), maxChannelNameLength) + } +} + +// TestResolveChannelSelectionRefusesPluginsChannel pins that the host-to-plugin +// dispatch channel is not reachable from a client connection, for any role. +func TestResolveChannelSelectionRefusesPluginsChannel(t *testing.T) { + for _, role := range []string{"user", "admin"} { + t.Run(role, func(t *testing.T) { + subs, accepted, rejected := resolveChannelSelection( + []evt.EventChannel{evt.ChannelPlugins}, allowedChannelsForRole(role), "profile-1") + + if len(subs) != 0 || len(accepted) != 0 { + t.Fatalf("%s was granted the plugins channel: %v", role, accepted) + } + if len(rejected) != 1 || rejected[0].Code != "unknown_channel" { + t.Errorf("rejected = %v, want a single unknown_channel entry", rejected) + } + }) + } +} diff --git a/internal/api/router.go b/internal/api/router.go index 33f3281c..e530e32b 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -1996,6 +1996,7 @@ func NewRouter(deps Dependencies) chi.Router { ) eventsHandler.SetNotificationsSystem(deps.Notifications) r.Get("/events/ws", eventsHandler.HandleWebSocket) + r.Get("/events/capability", eventsHandler.HandleCapability) } // User notifications: profile-scoped inbox, preferences, and diff --git a/internal/events/types.go b/internal/events/types.go index 1921422f..fed5330a 100644 --- a/internal/events/types.go +++ b/internal/events/types.go @@ -38,6 +38,24 @@ var AllChannels = []EventChannel{ ChannelNotifications, } +// ClientChannels is every channel a websocket client may subscribe to: it is +// AllChannels minus ChannelPlugins, which carries host-to-plugin runtime +// dispatch and is granted to no role, not even admin. Naming it in a +// capability response or accepting it as a valid subscription target would +// point a client at a request that can never succeed. +var ClientChannels = []EventChannel{ + ChannelCatalog, + ChannelJobs, + ChannelSessions, + ChannelTasks, + ChannelScans, + ChannelHistoryImport, + ChannelUserState, + ChannelUserSettings, + ChannelSettings, + ChannelNotifications, +} + type Envelope struct { Channel EventChannel `json:"channel"` Event string `json:"event"` diff --git a/web/src/api/types.ts b/web/src/api/types.ts index 968b459b..d87489f6 100644 --- a/web/src/api/types.ts +++ b/web/src/api/types.ts @@ -2895,7 +2895,14 @@ export interface EventsHelloMessage { schema_version: number; connection_id: string; available_channels: EventChannel[]; - required_action: "subscribe"; + /** + * "none" when the connection already holds at least one subscription, + * declared as ?channels= on the URL. "subscribe" when it still owes a + * subscribe frame — including when it declared channels but none of them + * resolved, since such a connection is subscribed to nothing and is closed + * after the grace period like any other silent one. + */ + required_action: "subscribe" | "none"; } export interface EventsSubscribeMessage {