feat(events): let websocket clients declare channels on connect (#525)
* 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Quick
@@ -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,
|
||||
})
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user