From 36be31db43a49ea167e12f3108a028b365b4efe5 Mon Sep 17 00:00:00 2001 From: J3vb <192430104+J3vb@users.noreply.github.com> Date: Sun, 16 Aug 2026 17:06:11 +0200 Subject: [PATCH] fix: 11 defects from bughunt sweep across server, db and client (#1382) * fix(ws): 1 defect(s) (OC-0001) * fix(db): 1 defect(s) (OC-0002) sanitizeFTSQuery filtered only characters, so FTS5's bareword boolean keywords (AND, OR, NOT) reached MATCH as operators; a query in an invalid operator position raised "fts5: syntax error" instead of returning results. Drop those bareword tokens after sanitizing. * fix(dm): 1 defect(s) (OC-0004) * fix(ws): 1 defect(s) (OC-0005) * fix(voice): 1 defect(s) (OC-0006) Count the shared voice_max_video budget in streams rather than rows: a single user publishing both camera and screenshare consumed one slot while producing two live streams, letting a channel over-admit up to 2N streams against an N-stream cap. * fix(client): 2 defect(s) (OC-0007, OC-0009) OC-0007: mark the active channel loading before invalidating its message window on a full-ready resync, so MessageList shows the spinner instead of the empty-channel state for the duration of the refetch. OC-0009: fan USER_UPDATE renames out to voiceStore.voiceUsers, which keeps its own frozen username copy, so the voice roster no longer shows a stale name for the rest of the call. * fix(admin): 1 defect(s) (OC-0010) * fix(identity): 1 defect(s) (OC-0011) * fix(ws): 1 defect(s) (OC-0003) The public half of an invisible user's presence (PresenceOthersEvent, and BroadcastPresence's own mapped payload) went out via broadcastExcludeLow on the low-priority queue - the ephemeral, unsequenced, drop-on-overflow transport built for typing indicators - while every other source of the same user's presence shares the normal-priority queue. That split one user's presence across two per-client FIFOs with different durability and different drain order (writePump drains normal strictly before low), so a frame could land out of order against a later connect/disconnect presence frame, or be silently dropped with no replay recovery. Adds Hub.BroadcastToAllExcept, which routes through the same h.broadcast channel and seqMu-serialized deliverBroadcast as BroadcastToAll, carrying an excludeUserID that deliverBroadcast applies via pubsub.Publish(TopicGlobal, msg, excludeUserID). * fix(ws): 1 defect(s) (OC-0008) --------- Co-authored-by: Claude --- Client/tauri-client/src/lib/dispatcher.ts | 14 ++ Client/tauri-client/src/stores/voice.store.ts | 22 +++ .../tests/unit/dispatcher.test.ts | 86 ++++++++ Server/admin/api_test.go | 54 +++++ Server/admin/handlers_channels.go | 24 ++- Server/api/totp_handler.go | 66 ++++++- Server/api/totp_handler_test.go | 144 ++++++++++++++ Server/db/dbgen/querier.go | 12 +- Server/db/dbgen/voice.sql.go | 24 ++- Server/db/message_queries.go | 20 +- Server/db/message_queries_test.go | 36 ++++ Server/db/queries/sqlite/voice.sql | 16 +- Server/db/voice_queries.go | 4 +- Server/service/dm.go | 15 +- Server/service/dm_test.go | 60 ++++++ Server/ws/emit.go | 45 ++++- .../ws/emit_presence_others_priority_test.go | 96 +++++++++ Server/ws/event.go | 8 +- Server/ws/hub_broadcast.go | 114 +++++++++-- Server/ws/oc_0006_video_stream_count_test.go | 119 +++++++++++ .../ws/presence_coalesce_flush_race_test.go | 97 +++++++++ .../reconnect_fallback_channel_leak_test.go | 184 ++++++++++++++++++ Server/ws/serve.go | 14 ++ Server/ws/serve_ready.go | 32 +++ Server/ws/serve_ready_dm_status_test.go | 102 ++++++++++ 25 files changed, 1346 insertions(+), 62 deletions(-) create mode 100644 Server/ws/emit_presence_others_priority_test.go create mode 100644 Server/ws/oc_0006_video_stream_count_test.go create mode 100644 Server/ws/presence_coalesce_flush_race_test.go create mode 100644 Server/ws/reconnect_fallback_channel_leak_test.go create mode 100644 Server/ws/serve_ready_dm_status_test.go diff --git a/Client/tauri-client/src/lib/dispatcher.ts b/Client/tauri-client/src/lib/dispatcher.ts index 7db8941b..7efa1768 100644 --- a/Client/tauri-client/src/lib/dispatcher.ts +++ b/Client/tauri-client/src/lib/dispatcher.ts @@ -29,6 +29,7 @@ import { messagesStore, setMessages, invalidateLoadedMessageWindows, + setChannelLoading, setChannelLoadError, } from "@stores/messages.store"; import { @@ -44,6 +45,7 @@ import { voiceStore, setVoiceStates, updateVoiceState, + updateVoiceUserProfile, removeVoiceUser, setVoiceConfig, setSpeakers, @@ -364,6 +366,13 @@ export function wireDispatcher( // showing only carried-through pending rows until the user navigates // away and back. if (activeAfterReady !== null && getMessages !== undefined) { + // Mark the active channel loading BEFORE invalidating its window — + // invalidate drops its rows synchronously, and if historyLoadState + // is left idle for even one microtask, MessageList's "no rows + + // idle" empty-state branch renders the channel as genuinely empty + // for the whole in-flight refetch instead of showing the in-region + // spinner (OC-0007). + setChannelLoading(activeAfterReady); invalidateLoadedMessageWindows(); getMessages(activeAfterReady, { limit: 50 }) .then((resp) => { @@ -797,6 +806,11 @@ export function wireDispatcher( avatar: payload.avatar ?? "", ...(payload.display_name === undefined ? {} : { displayName: payload.display_name ?? "" }), }); + // voiceStore.voiceUsers is the third store holding a frozen username + // copy (see updateVoiceUserProfile's doc comment) — without this, a + // rename leaves the voice roster showing the old name for the rest of + // the call. + updateVoiceUserProfile(payload.user_id, { username: payload.username }); // Update auth store if the current user changed their own profile. const currentUser = authStore.getState().user; diff --git a/Client/tauri-client/src/stores/voice.store.ts b/Client/tauri-client/src/stores/voice.store.ts index 07cbc6e7..11db02b8 100644 --- a/Client/tauri-client/src/stores/voice.store.ts +++ b/Client/tauri-client/src/stores/voice.store.ts @@ -244,6 +244,28 @@ export function updateVoiceState(payload: VoiceStatePayload): void { }); } +/** Patch a user's username in every voice channel they currently occupy. + * VoiceUser keeps its own frozen copy of the username (the same shape as + * membersStore's and dmStore's copies), refreshed only by setVoiceStates + * (ready) and updateVoiceState (voice_state) — a profile rename via + * USER_UPDATE must patch it too, or the voice roster keeps showing the old + * name for the rest of the call, until the user leaves and rejoins voice. */ +export function updateVoiceUserProfile(userId: number, patch: { readonly username: string }): void { + voiceStore.setState((prev) => { + let changed = false; + const nextChannels = new Map(prev.voiceUsers); + for (const [channelId, users] of prev.voiceUsers) { + const existing = users.get(userId); + if (existing === undefined) continue; + const nextUsers = new Map(users); + nextUsers.set(userId, { ...existing, username: patch.username }); + nextChannels.set(channelId, nextUsers); + changed = true; + } + return changed ? { ...prev, voiceUsers: nextChannels } : prev; + }); +} + /** Remove a user from a voice channel. */ export function removeVoiceUser(payload: VoiceLeavePayload): void { voiceStore.setState((prev) => { diff --git a/Client/tauri-client/tests/unit/dispatcher.test.ts b/Client/tauri-client/tests/unit/dispatcher.test.ts index 071747fe..5ad20f28 100644 --- a/Client/tauri-client/tests/unit/dispatcher.test.ts +++ b/Client/tauri-client/tests/unit/dispatcher.test.ts @@ -695,6 +695,44 @@ describe("WS Dispatcher", () => { expect(member?.avatar).toBe("/api/v1/files/abc"); }); + // OC-0009: voiceStore.voiceUsers keeps its own frozen copy of each + // participant's username, written only by setVoiceStates (ready) and + // updateVoiceState (voice_state). USER_UPDATE fans a profile change out to + // membersStore and dmStore, which hold the same kind of frozen copy, but + // not to voiceStore — so a rename leaves the voice roster showing the old + // username for the rest of the call. + it("updates the voice roster username on user_update", () => { + voiceStore.setState((prev) => { + const userMap = new Map([ + [ + 10, + { + userId: 10, + username: "bob", + muted: false, + deafened: false, + speaking: false, + camera: false, + screenshare: false, + }, + ], + ]); + const voiceUsers = new Map(prev.voiceUsers); + voiceUsers.set(5, userMap); + return { ...prev, voiceUsers }; + }); + + mock.dispatch("user_update", { + user_id: 10, + username: "bobby", + avatar: "new.png", + display_name: null, + }); + + const voiceUser = voiceStore.getState().voiceUsers.get(5)?.get(10); + expect(voiceUser?.username).toBe("bobby"); + }); + describe("presence and user_update sync dmStore", () => { const dmChannel = { channelId: 50, @@ -1378,6 +1416,54 @@ describe("WS Dispatcher", () => { expect(isChannelLoaded(1)).toBe(true); }); + // OC-0007: invalidateLoadedMessageWindows() synchronously drops the + // active channel's rows and loadedChannels entry, but leaves + // historyLoadState untouched. If the resync handler doesn't also mark + // the channel "loading", MessageList's empty-state branch (no rows + + // idle load state) reads as a genuinely empty channel for the entire + // round trip of the refetch, instead of showing the in-region spinner + // every other first-page fetch shows. + it("marks the active channel loading (not idle) while the resync refetch is in flight", () => { + cleanup(); + const listBlocks = vi.fn().mockResolvedValue({ blocked_user_ids: [] }); + // Never resolves within this test — pins the synchronous state right + // after the resync `ready` is handled, before the refetch lands. + const getMessages = vi.fn().mockReturnValue(new Promise(() => {})); + cleanup = wireDispatcher(mock.ws, { listBlocks, getMessages }); + + channelsStore.setState((prev) => ({ ...prev, activeChannelId: 1 })); + setMessages(1, [storedMessage(10)], false); + const readyChannels = [ + { id: 1, name: "general", type: "text" as const, category: null, position: 0 }, + ]; + + // First ready: initial connect. + mock.dispatch("ready", { + channels: readyChannels, + members: [], + voice_states: [], + roles: [], + dm_channels: [], + }); + + // Second ready: a full-ready resync. The synchronous invalidation runs + // and the refetch is fired but never resolves in this test. + mock.dispatch("ready", { + channels: readyChannels, + members: [], + voice_states: [], + roles: [], + dm_channels: [], + }); + + // The window was dropped... + expect(isChannelLoaded(1)).toBe(false); + expect(getChannelMessages(1)).toEqual([]); + // ...but the load state must be "loading", not idle (null) — idle+empty + // is exactly the shape MessageList reads as "channel has no history". + expect(getHistoryLoadState(1)).toBe("loading"); + }); + // OC-0203: the refetch above is fired-and-forgotten against whatever // channel was active when the resync `ready` arrived — but the user can // switch channels before the HTTP response lands. The continuation must diff --git a/Server/admin/api_test.go b/Server/admin/api_test.go index 0e478790..0e4981db 100644 --- a/Server/admin/api_test.go +++ b/Server/admin/api_test.go @@ -1436,6 +1436,60 @@ func TestAdminAPI_DeleteChannel_NilHubDoesNotPanic(t *testing.T) { } } +// OC-0010: handleDeleteChannel commits archived=1 as its own transaction and +// evicts voice participants BEFORE calling AdminDeleteChannel — all on +// r.Context(). If the admin's browser aborts the request in that window (tab +// close, navigation, network blip), r.Context() is canceled and the final +// AdminDeleteChannel call fails with context.Canceled: the handler 500s, but +// the archive and the voice eviction already committed. Nothing reverts the +// archive, nothing broadcasts it, and no audit row is written — the channel +// is left silently archived (writes refused, channel_focus 403s) while every +// connected client still shows it live in the sidebar. +// +// This reproduces the race deterministically by canceling the request +// context from inside CleanupVoiceForChannel — exactly the call the repro +// says the real-world cancellation lands during — instead of relying on +// timing. The fix must make the delete tolerate a caller cancellation that +// arrives after the archive has already committed. +func TestAdminAPI_DeleteChannel_SurvivesContextCancelAfterArchiveCommits(t *testing.T) { + database := openAdminTestDB(t) + hub := &mockHub{} + handler := admin.NewAdminAPI(database, "1.0.0", hub, nil, nil, nil, nil, newTestModService(database), newTestRoleService(database)) + token := createAdminUser(t, database) + + chID, _ := database.AdminCreateChannel(context.Background(), "del-cancel-race", "text", "", "", 0) + + ctx, cancel := context.WithCancel(context.Background()) + // Fires synchronously inside handleDeleteChannel, after the archive + // commit but before the final AdminDeleteChannel call — the same window + // the repro describes a browser abort landing in. + hub.onVoiceCleanup = func(int64) { + cancel() + } + + req := httptest.NewRequest(http.MethodDelete, "/channels/"+itoa(chID), nil) + req = req.WithContext(ctx) + req.Header.Set("Authorization", "Bearer "+token) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + if w.Code != http.StatusNoContent { + t.Fatalf("status = %d, want 204 (delete must survive a caller cancellation that arrives after the archive already committed); body: %s", w.Code, w.Body.String()) + } + + ch, err := database.GetChannel(context.Background(), chID) + if err != nil { + t.Fatalf("GetChannel: %v", err) + } + if ch != nil { + t.Errorf("channel %d still exists after a reported-successful delete: %+v", chID, ch) + } + + if len(hub.channelDeleteIDs) != 1 || hub.channelDeleteIDs[0] != chID { + t.Errorf("BroadcastChannelDelete calls = %v, want exactly [%d]", hub.channelDeleteIDs, chID) + } +} + // ─── API tokens: /admin/api/tokens ─────────────────────────────────────────── func TestAdminAPI_CreateAPIToken_OK(t *testing.T) { diff --git a/Server/admin/handlers_channels.go b/Server/admin/handlers_channels.go index 3edb8bef..2c979797 100644 --- a/Server/admin/handlers_channels.go +++ b/Server/admin/handlers_channels.go @@ -317,13 +317,33 @@ func handleDeleteChannel(database *db.DB, hub HubBroadcaster) http.HandlerFunc { hub.CleanupVoiceForChannel(id) } - if err := database.AdminDeleteChannel(r.Context(), id); err != nil { + // From here on the archive (and, if it ran, the voice eviction) + // already committed. If the admin's browser goes away in this window + // (tab close, navigation, network blip), r.Context() cancels, and an + // AdminDeleteChannel that still used it would fail with + // context.Canceled — 500ing while leaving the channel silently + // archived, unbroadcast and unaudited (OC-0010). Run the rest of the + // delete on an uncancellable tail, matching the repo's convention + // for other durable side effects (totp_handler.go, service/user.go). + delCtx := context.WithoutCancel(r.Context()) + if err := database.AdminDeleteChannel(delCtx, id); err != nil { + // A genuine failure here (not caller cancellation, which delCtx + // already absorbs) still leaves the archive committed — tell + // connected clients about the state that did change instead of + // leaving them stuck seeing a live channel that now 403s on + // every read and write. + if hub != nil && !existing.Archived { + if archived, gErr := database.GetChannel(delCtx, id); gErr == nil && archived != nil { + hub.BroadcastChannelUpdate(archived) + hub.RefreshChannelVisibility(archived) + } + } writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to delete channel") return } actor := actorFromContext(r) slog.Warn("channel deleted", "actor_id", actor, "channel_id", id, "name", existing.Name) - db.WriteAudit(context.WithoutCancel(r.Context()), database, actor, "channel_delete", "channel", id, + db.WriteAudit(delCtx, database, actor, "channel_delete", "channel", id, fmt.Sprintf("deleted #%s", existing.Name)) if hub != nil { hub.BroadcastChannelDelete(id) diff --git a/Server/api/totp_handler.go b/Server/api/totp_handler.go index c64b9619..1b168d67 100644 --- a/Server/api/totp_handler.go +++ b/Server/api/totp_handler.go @@ -225,6 +225,30 @@ func handleEnableTOTP(pendingStore *auth.PendingTOTPStore, limiter *auth.RateLim } } +// revokeOtherSessionsAfterAuthChange revokes every session for userID except +// keepSessionID as the security tail of a committed 2FA state change. It +// mirrors UserService.ChangePassword (service/user.go:262-274): a failure is +// logged and retried once (bounded compensating retry for transient write +// contention); if the retry also fails, revoked reports what did succeed and +// failed is true so the caller can report a partial success instead of +// silently claiming the other sessions were revoked when they were not. +func revokeOtherSessionsAfterAuthChange(ctx context.Context, database *db.DB, userID, keepSessionID int64, action string) (revoked int64, failed bool) { + revoked, err := database.DeleteOtherSessions(ctx, userID, keepSessionID) + if err != nil { + slog.Error("DeleteOtherSessions after "+action, "err", err, "user_id", userID) + revokedRetry, retryErr := database.DeleteOtherSessions(ctx, userID, keepSessionID) + if retryErr != nil { + slog.Error("DeleteOtherSessions retry after "+action, "err", retryErr, "user_id", userID) + return revoked, true + } + revoked += revokedRetry + } + if revoked > 0 { + slog.Info("revoked other sessions after "+action, "user_id", userID, "revoked", revoked) + } + return revoked, false +} + func handleConfirmTOTP(database *db.DB, pendingStore *auth.PendingTOTPStore, usedTOTPCodes *auth.UsedTOTPCodeStore, limiter *auth.RateLimiter, totpKey []byte) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { user, ok := r.Context().Value(UserKey).(*db.User) @@ -313,15 +337,26 @@ func handleConfirmTOTP(database *db.DB, pendingStore *auth.PendingTOTPStore, use } // Security tail of the 2FA change: once the secret update committed, // revoking the other sessions must not be aborted by a dead request. - n, _ := database.DeleteOtherSessions(context.WithoutCancel(r.Context()), user.ID, keepSessionID) - if n > 0 { - slog.Info("revoked other sessions after totp enable", "user_id", user.ID, "revoked", n) - } + tailCtx := context.WithoutCancel(r.Context()) + revoked, revokeFailed := revokeOtherSessionsAfterAuthChange(tailCtx, database, user.ID, keepSessionID, "totp enable") slog.Info("totp enabled", "user_id", user.ID) - db.WriteAudit(context.WithoutCancel(r.Context()), database, user.ID, "totp_enabled", "user", user.ID, + db.WriteAudit(tailCtx, database, user.ID, "totp_enabled", "user", user.ID, "two-factor authentication enrolled") + if revokeFailed { + // Partial success: 2FA IS enabled; only revoking the other + // sessions failed. A 5xx here would be a lie — the state change + // already committed — so mirror the ChangePassword contract + // (api/profile_handler.go) and report 200 with an explicit warning + // instead of a silent, unqualified 204. + writeJSON(w, http.StatusOK, map[string]any{ + "warning": "two-factor authentication enabled, but other sessions could not be revoked; revoke them from the sessions list", + "sessions_revoked": revoked, + }) + return + } + w.WriteHeader(http.StatusNoContent) } } @@ -403,15 +438,26 @@ func handleDisableTOTP(database *db.DB, pendingStore *auth.PendingTOTPStore, lim } // Security tail of the 2FA change: once the secret update committed, // revoking the other sessions must not be aborted by a dead request. - n, _ := database.DeleteOtherSessions(context.WithoutCancel(r.Context()), user.ID, keepSessionID) - if n > 0 { - slog.Info("revoked other sessions after totp disable", "user_id", user.ID, "revoked", n) - } + tailCtx := context.WithoutCancel(r.Context()) + revoked, revokeFailed := revokeOtherSessionsAfterAuthChange(tailCtx, database, user.ID, keepSessionID, "totp disable") slog.Info("totp disabled", "user_id", user.ID) - db.WriteAudit(context.WithoutCancel(r.Context()), database, user.ID, "totp_disabled", "user", user.ID, + db.WriteAudit(tailCtx, database, user.ID, "totp_disabled", "user", user.ID, "two-factor authentication disabled") + if revokeFailed { + // Partial success: 2FA IS disabled; only revoking the other + // sessions failed. A 5xx here would be a lie — the state change + // already committed — so mirror the ChangePassword contract + // (api/profile_handler.go) and report 200 with an explicit warning + // instead of a silent, unqualified 204. + writeJSON(w, http.StatusOK, map[string]any{ + "warning": "two-factor authentication disabled, but other sessions could not be revoked; revoke them from the sessions list", + "sessions_revoked": revoked, + }) + return + } + w.WriteHeader(http.StatusNoContent) } } diff --git a/Server/api/totp_handler_test.go b/Server/api/totp_handler_test.go index 5c270791..d5864133 100644 --- a/Server/api/totp_handler_test.go +++ b/Server/api/totp_handler_test.go @@ -528,6 +528,150 @@ func TestDisableTOTP_APITokenPrincipal_RevokesAllSessions(t *testing.T) { } } +// ─── OC-0011: DeleteOtherSessions failures must not be silent successes ────── + +// TestConfirmTOTP_RevokeFailureSurfacesWarning locks down that when +// DeleteOtherSessions fails after a 2FA enable commits, the handler must not +// report unqualified success: it should mirror the ChangePassword contract +// (service/user.go:262-274 / api/profile_handler.go:377) and report a 200 +// with an explicit warning, not a silent 204. +func TestConfirmTOTP_RevokeFailureSurfacesWarning(t *testing.T) { + database := newAuthTestDB(t) + limiter := auth.NewRateLimiter() + router := buildAuthRouter(database, limiter) + + token := loginAndGetToken(t, router, database, "confirmrevokefail", 4) + user, _ := database.GetUserByUsername(context.Background(), "confirmrevokefail") + if user == nil { + t.Fatal("user not found") + } + + // A second, unrelated session so DeleteOtherSessions has a row to delete + // (a DELETE that matches zero rows never fires a BEFORE DELETE trigger). + otherToken, _ := auth.GenerateToken() + if _, err := database.CreateSession(context.Background(), user.ID, auth.HashToken(otherToken), "other-device", "127.0.0.1"); err != nil { + t.Fatalf("CreateSession: %v", err) + } + + // Step 1: enable to get a pending secret (before the trigger exists). + rr := postJSONWithToken(t, router, "/api/v1/users/me/totp/enable", token, + map[string]string{"password": "Password1!"}) + if rr.Code != http.StatusOK { + t.Fatalf("enable: status = %d; body = %s", rr.Code, rr.Body.String()) + } + var enableResp map[string]any + _ = json.NewDecoder(rr.Body).Decode(&enableResp) + secret := extractSecretFromURI(t, enableResp["qr_uri"].(string)) + code, _ := auth.GenerateTOTPCode(secret, time.Now().UTC()) + + // Make every DELETE against sessions fail from here on, simulating a + // genuine DB-level failure (disk error, "database is closed" during + // shutdown) rather than a cancel race — the confirm call below already + // uses context.WithoutCancel, so a dead request cannot trigger this. + if _, err := database.ExecContext(context.Background(), ` + CREATE TRIGGER block_delete_sessions + BEFORE DELETE ON sessions + BEGIN + SELECT RAISE(FAIL, 'delete blocked'); + END; + `); err != nil { + t.Fatalf("create trigger: %v", err) + } + + // Step 2: confirm. The secret update must still commit even though the + // session-revocation tail fails. + rr = postJSONWithToken(t, router, "/api/v1/users/me/totp/confirm", token, + map[string]string{"password": "Password1!", "code": code}) + + updated, _ := database.GetUserByUsername(context.Background(), "confirmrevokefail") + if updated == nil || updated.TOTPSecret == nil { + t.Fatal("expected TOTPSecret to be committed even though revocation failed") + } + + if rr.Code != http.StatusOK { + t.Fatalf("confirm-totp with revoke failure: status = %d, want 200; body = %s", rr.Code, rr.Body.String()) + } + var resp map[string]any + if err := json.NewDecoder(rr.Body).Decode(&resp); err != nil { + t.Fatalf("decode confirm response: %v", err) + } + if resp["warning"] == nil { + t.Error("expected a warning field when session revocation failed after totp enable") + } + + // The other session must still be present — revocation genuinely failed, + // it was not silently skipped or falsely reported as revoked. + if s, _ := database.GetSessionByTokenHash(context.Background(), auth.HashToken(otherToken)); s == nil { + t.Error("other session should still exist: DeleteOtherSessions was blocked by the trigger") + } +} + +// TestDisableTOTP_RevokeFailureSurfacesWarning is the handleDisableTOTP +// sibling of TestConfirmTOTP_RevokeFailureSurfacesWarning. +func TestDisableTOTP_RevokeFailureSurfacesWarning(t *testing.T) { + database := newAuthTestDB(t) + limiter := auth.NewRateLimiter() + router := buildAuthRouter(database, limiter) + + token := loginAndGetToken(t, router, database, "disablerevokefail", 4) + user, _ := database.GetUserByUsername(context.Background(), "disablerevokefail") + if user == nil { + t.Fatal("user not found") + } + + // Enable and confirm TOTP first (before the trigger exists). + rr := postJSONWithToken(t, router, "/api/v1/users/me/totp/enable", token, + map[string]string{"password": "Password1!"}) + var enableResp map[string]any + _ = json.NewDecoder(rr.Body).Decode(&enableResp) + secret := extractSecretFromURI(t, enableResp["qr_uri"].(string)) + code, _ := auth.GenerateTOTPCode(secret, time.Now().UTC()) + rr = postJSONWithToken(t, router, "/api/v1/users/me/totp/confirm", token, + map[string]string{"password": "Password1!", "code": code}) + if rr.Code != http.StatusNoContent { + t.Fatalf("setup confirm: status = %d; body = %s", rr.Code, rr.Body.String()) + } + + // A second, unrelated session so DeleteOtherSessions has a row to delete. + otherToken, _ := auth.GenerateToken() + if _, err := database.CreateSession(context.Background(), user.ID, auth.HashToken(otherToken), "other-device", "127.0.0.1"); err != nil { + t.Fatalf("CreateSession: %v", err) + } + + if _, err := database.ExecContext(context.Background(), ` + CREATE TRIGGER block_delete_sessions + BEFORE DELETE ON sessions + BEGIN + SELECT RAISE(FAIL, 'delete blocked'); + END; + `); err != nil { + t.Fatalf("create trigger: %v", err) + } + + rr = deleteWithToken(t, router, "/api/v1/users/me/totp", token, + map[string]string{"password": "Password1!"}) + + updated, _ := database.GetUserByUsername(context.Background(), "disablerevokefail") + if updated == nil || updated.TOTPSecret != nil { + t.Fatal("expected TOTPSecret to be cleared even though revocation failed") + } + + if rr.Code != http.StatusOK { + t.Fatalf("disable-totp with revoke failure: status = %d, want 200; body = %s", rr.Code, rr.Body.String()) + } + var resp map[string]any + if err := json.NewDecoder(rr.Body).Decode(&resp); err != nil { + t.Fatalf("decode disable response: %v", err) + } + if resp["warning"] == nil { + t.Error("expected a warning field when session revocation failed after totp disable") + } + + if s, _ := database.GetSessionByTokenHash(context.Background(), auth.HashToken(otherToken)); s == nil { + t.Error("other session should still exist: DeleteOtherSessions was blocked by the trigger") + } +} + // ─── Helpers ───────────────────────────────────────────────────────────────── // deleteWithToken sends a DELETE request with a JSON body and auth token. diff --git a/Server/db/dbgen/querier.go b/Server/db/dbgen/querier.go index 4e88f3c9..f4b220c9 100644 --- a/Server/db/dbgen/querier.go +++ b/Server/db/dbgen/querier.go @@ -70,10 +70,14 @@ type Querier interface { DeleteSessionByToken(ctx context.Context, token string) error DisablePlugin(ctx context.Context, id int64) error EditMessageContent(ctx context.Context, arg EditMessageContentParams) (Message, error) - // Camera and screenshare share one voice_max_video budget: a channel capped - // at N simultaneous video streams must not let a camera publish ignore - // screenshare occupants (or vice versa), so both gates count the same - // `camera = 1 OR screenshare = 1` slot usage (OC-0023). + // Camera and screenshare share one voice_max_video budget, counted in + // STREAMS, not rows: a channel capped at N simultaneous video streams must + // not let a camera publish ignore screenshare occupants (or vice versa, + // OC-0023), and a single user with both flags set must consume two of the N + // slots, not one (OC-0006) -- so both gates sum `vs2.camera + vs2.screenshare` + // across the channel's rows rather than counting rows where either is set. + // The enabling user's own bit is still 0 at gate time, so no self-exclusion + // term is needed. EnableCameraIfUnderLimit(ctx context.Context, arg EnableCameraIfUnderLimitParams) (sql.Result, error) EnablePlugin(ctx context.Context, id int64) error EnableScreenshareIfUnderLimit(ctx context.Context, arg EnableScreenshareIfUnderLimitParams) (sql.Result, error) diff --git a/Server/db/dbgen/voice.sql.go b/Server/db/dbgen/voice.sql.go index a8dfb8f9..708ef0b7 100644 --- a/Server/db/dbgen/voice.sql.go +++ b/Server/db/dbgen/voice.sql.go @@ -102,40 +102,44 @@ const enableCameraIfUnderLimit = `-- name: EnableCameraIfUnderLimit :execresult UPDATE voice_states SET camera = 1 WHERE voice_states.user_id = ? AND voice_states.channel_id = ? - AND (SELECT COUNT(*) FROM voice_states AS vs2 WHERE vs2.channel_id = ? AND (vs2.camera = 1 OR vs2.screenshare = 1)) < ? + AND (SELECT COALESCE(SUM(vs2.camera), 0) + COALESCE(SUM(vs2.screenshare), 0) FROM voice_states AS vs2 WHERE vs2.channel_id = ?) < ?4 ` type EnableCameraIfUnderLimitParams struct { UserID int64 `json:"userId"` ChannelID int64 `json:"channelId"` ChannelID_2 int64 `json:"channelId2"` - ChannelID_3 int64 `json:"channelId3"` + MaxVideo int64 `json:"maxVideo"` } -// Camera and screenshare share one voice_max_video budget: a channel capped -// at N simultaneous video streams must not let a camera publish ignore -// screenshare occupants (or vice versa), so both gates count the same -// `camera = 1 OR screenshare = 1` slot usage (OC-0023). +// Camera and screenshare share one voice_max_video budget, counted in +// STREAMS, not rows: a channel capped at N simultaneous video streams must +// not let a camera publish ignore screenshare occupants (or vice versa, +// OC-0023), and a single user with both flags set must consume two of the N +// slots, not one (OC-0006) -- so both gates sum `vs2.camera + vs2.screenshare` +// across the channel's rows rather than counting rows where either is set. +// The enabling user's own bit is still 0 at gate time, so no self-exclusion +// term is needed. func (q *Queries) EnableCameraIfUnderLimit(ctx context.Context, arg EnableCameraIfUnderLimitParams) (sql.Result, error) { return q.db.ExecContext(ctx, enableCameraIfUnderLimit, arg.UserID, arg.ChannelID, arg.ChannelID_2, - arg.ChannelID_3, + arg.MaxVideo, ) } const enableScreenshareIfUnderLimit = `-- name: EnableScreenshareIfUnderLimit :execresult UPDATE voice_states SET screenshare = 1 WHERE voice_states.user_id = ? AND voice_states.channel_id = ? - AND (SELECT COUNT(*) FROM voice_states AS vs2 WHERE vs2.channel_id = ? AND (vs2.camera = 1 OR vs2.screenshare = 1)) < ? + AND (SELECT COALESCE(SUM(vs2.camera), 0) + COALESCE(SUM(vs2.screenshare), 0) FROM voice_states AS vs2 WHERE vs2.channel_id = ?) < ?4 ` type EnableScreenshareIfUnderLimitParams struct { UserID int64 `json:"userId"` ChannelID int64 `json:"channelId"` ChannelID_2 int64 `json:"channelId2"` - ChannelID_3 int64 `json:"channelId3"` + MaxVideo int64 `json:"maxVideo"` } func (q *Queries) EnableScreenshareIfUnderLimit(ctx context.Context, arg EnableScreenshareIfUnderLimitParams) (sql.Result, error) { @@ -143,7 +147,7 @@ func (q *Queries) EnableScreenshareIfUnderLimit(ctx context.Context, arg EnableS arg.UserID, arg.ChannelID, arg.ChannelID_2, - arg.ChannelID_3, + arg.MaxVideo, ) } diff --git a/Server/db/message_queries.go b/Server/db/message_queries.go index 14efa48f..a8620f4b 100644 --- a/Server/db/message_queries.go +++ b/Server/db/message_queries.go @@ -34,6 +34,15 @@ func messageFromGen(m dbgen.Message) *Message { // ("-col: expr"), so keeping it turns "well-known" into a filter on a // nonexistent column "known" and SQLite errors instead of matching. Folding // to a space (rather than dropping it) still matches the indexed tokens. +// +// Filtering characters alone is not enough: FTS5's MATCH grammar also +// recognizes bareword keywords -- AND, OR, NOT (uppercase only) -- as +// boolean operators rather than search terms. Those are ordinary letters, so +// the character filter lets them through unchanged, and a query that places +// one in an invalid position (e.g. the whole query is "AND", or it starts or +// ends with one) makes SQLite raise "fts5: syntax error" instead of running +// the search. Any such token is dropped below so the result is always a +// plain sequence of bareword terms. func sanitizeFTSQuery(q string) string { var sb strings.Builder sb.Grow(len(q)) @@ -51,7 +60,16 @@ func sanitizeFTSQuery(q string) string { if runes := []rune(result); len(runes) > 200 { result = string(runes[:200]) } - return result + + fields := strings.Fields(result) + kept := fields[:0] + for _, f := range fields { + if f == "AND" || f == "OR" || f == "NOT" { + continue + } + kept = append(kept, f) + } + return strings.Join(kept, " ") } // CreateMessage inserts a new message and returns the assigned ID. diff --git a/Server/db/message_queries_test.go b/Server/db/message_queries_test.go index 101f6b33..aa373c46 100644 --- a/Server/db/message_queries_test.go +++ b/Server/db/message_queries_test.go @@ -720,6 +720,42 @@ func TestSearchMessages_HyphenatedQuery(t *testing.T) { } } +// OC-0002: sanitizeFTSQuery only filters characters, not FTS5's bareword +// boolean keywords (AND, OR, NOT). A query consisting of (or containing) one +// of those keywords in an operator position makes SQLite raise an +// "fts5: syntax error", which SearchMessages surfaces as an error (mapped by +// the service layer to a 500) instead of returning zero results. +func TestSearchMessages_BooleanKeywordQuery(t *testing.T) { + database := openMigratedMemory(t) + userID := seedUser(t, database, "booleanuser") + chID := seedChannel(t, database, "booleanch") + + _, _ = database.CreateMessage(context.Background(), chID, userID, "hello world", nil) + + for _, q := range []string{"AND", "OR", "NOT", "AND world", "hello AND"} { + if _, err := database.SearchMessages(context.Background(), q, nil, 10); err != nil { + t.Errorf("SearchMessages(%q): unexpected error: %v", q, err) + } + } +} + +// Same root cause as TestSearchMessages_BooleanKeywordQuery, but through the +// SearchMessagesInChannels caller, which sanitizes with the same +// sanitizeFTSQuery helper. +func TestSearchMessagesInChannels_BooleanKeywordQuery(t *testing.T) { + database := openMigratedMemory(t) + userID := seedUser(t, database, "boolchanuser") + chID := seedChannel(t, database, "boolchanch") + + _, _ = database.CreateMessage(context.Background(), chID, userID, "hello world", nil) + + for _, q := range []string{"AND", "OR", "NOT", "AND world", "hello AND"} { + if _, err := database.SearchMessagesInChannels(context.Background(), q, []int64{chID}, 10); err != nil { + t.Errorf("SearchMessagesInChannels(%q): unexpected error: %v", q, err) + } + } +} + // ─── UpdateReadState ────────────────────────────────────────────────────────── func TestUpdateReadState_Upsert(t *testing.T) { diff --git a/Server/db/queries/sqlite/voice.sql b/Server/db/queries/sqlite/voice.sql index 3687da6d..66a7a5b3 100644 --- a/Server/db/queries/sqlite/voice.sql +++ b/Server/db/queries/sqlite/voice.sql @@ -93,20 +93,24 @@ UPDATE voice_states SET server_deafened = 1, deafened = 1 WHERE user_id = ? AND -- name: ClearVoiceServerDeafen :execresult UPDATE voice_states SET server_deafened = 0 WHERE user_id = ? AND channel_id = ?; --- Camera and screenshare share one voice_max_video budget: a channel capped --- at N simultaneous video streams must not let a camera publish ignore --- screenshare occupants (or vice versa), so both gates count the same --- `camera = 1 OR screenshare = 1` slot usage (OC-0023). +-- Camera and screenshare share one voice_max_video budget, counted in +-- STREAMS, not rows: a channel capped at N simultaneous video streams must +-- not let a camera publish ignore screenshare occupants (or vice versa, +-- OC-0023), and a single user with both flags set must consume two of the N +-- slots, not one (OC-0006) -- so both gates sum `vs2.camera + vs2.screenshare` +-- across the channel's rows rather than counting rows where either is set. +-- The enabling user's own bit is still 0 at gate time, so no self-exclusion +-- term is needed. -- name: EnableCameraIfUnderLimit :execresult UPDATE voice_states SET camera = 1 WHERE voice_states.user_id = ? AND voice_states.channel_id = ? - AND (SELECT COUNT(*) FROM voice_states AS vs2 WHERE vs2.channel_id = ? AND (vs2.camera = 1 OR vs2.screenshare = 1)) < ?; + AND (SELECT COALESCE(SUM(vs2.camera), 0) + COALESCE(SUM(vs2.screenshare), 0) FROM voice_states AS vs2 WHERE vs2.channel_id = ?) < sqlc.arg(max_video); -- name: EnableScreenshareIfUnderLimit :execresult UPDATE voice_states SET screenshare = 1 WHERE voice_states.user_id = ? AND voice_states.channel_id = ? - AND (SELECT COUNT(*) FROM voice_states AS vs2 WHERE vs2.channel_id = ? AND (vs2.camera = 1 OR vs2.screenshare = 1)) < ?; + AND (SELECT COALESCE(SUM(vs2.camera), 0) + COALESCE(SUM(vs2.screenshare), 0) FROM voice_states AS vs2 WHERE vs2.channel_id = ?) < sqlc.arg(max_video); -- name: ClearVoiceState :exec DELETE FROM voice_states WHERE user_id = ?; diff --git a/Server/db/voice_queries.go b/Server/db/voice_queries.go index 2684f5e6..37427a48 100644 --- a/Server/db/voice_queries.go +++ b/Server/db/voice_queries.go @@ -278,7 +278,7 @@ func (d *DB) EnableCameraIfUnderLimit(ctx context.Context, userID, channelID int UserID: userID, ChannelID: channelID, ChannelID_2: channelID, - ChannelID_3: int64(maxVideo), + MaxVideo: int64(maxVideo), }) if err != nil { return false, fmt.Errorf("EnableCameraIfUnderLimit: %w", err) @@ -311,7 +311,7 @@ func (d *DB) EnableScreenshareIfUnderLimit(ctx context.Context, userID, channelI UserID: userID, ChannelID: channelID, ChannelID_2: channelID, - ChannelID_3: int64(maxVideo), + MaxVideo: int64(maxVideo), }) if err != nil { return false, fmt.Errorf("EnableScreenshareIfUnderLimit: %w", err) diff --git a/Server/service/dm.go b/Server/service/dm.go index bd8a94a5..240cdbd1 100644 --- a/Server/service/dm.go +++ b/Server/service/dm.go @@ -273,9 +273,20 @@ func (s *DMService) CreateGroupDM(ctx context.Context, userID int64, recipientID return nil, fmt.Errorf("%w: failed to create group DM", ErrInternal) } - participants, err := s.st.GetDMParticipants(ctx, ch.ID, userID) + // The channel, all dm_participants rows and all dm_open_state rows are + // already committed at this point, so this read must not turn a fully- + // persisted group DM into a reported failure (OC-0004): it runs + // uncancellable (context.WithoutCancel) so a client disconnect landing + // right after the commit can't fail it via ctx.Err(), and any other + // failure is logged rather than propagated so the caller still gets a + // usable result to fan dm_channel_open out from. Group DMs are + // duplicate-by-design — there is no "the group for these users" to find — + // so telling the caller creation failed when it actually committed only + // invites a retry that creates a second, indistinguishable group. + participants, err := s.st.GetDMParticipants(context.WithoutCancel(ctx), ch.ID, userID) if err != nil { - return nil, fmt.Errorf("%w: failed to read group DM participants: %v", ErrInternal, err) + slog.Error("DMService.CreateGroupDM: failed to read participants after commit", "err", err, "channel_id", ch.ID) + participants = nil } return &CreateGroupDMResult{ diff --git a/Server/service/dm_test.go b/Server/service/dm_test.go index 626c1a21..2925ca71 100644 --- a/Server/service/dm_test.go +++ b/Server/service/dm_test.go @@ -61,3 +61,63 @@ func TestDMService_CreateGroupDM_RefusesBannedRecipient(t *testing.T) { t.Fatalf("CreateGroupDM with a banned recipient = %v, want ErrNotFound", err) } } + +// cancelAfterCreateGroupDMStore wraps a real *db.DB and cancels a context the +// instant CreateGroupDMChannel returns successfully — simulating a client +// disconnect that lands exactly in the gap between the channel's commit and +// the service's post-commit GetDMParticipants read. +type cancelAfterCreateGroupDMStore struct { + *db.DB + cancel context.CancelFunc +} + +func (s *cancelAfterCreateGroupDMStore) CreateGroupDMChannel(ctx context.Context, name string, participantIDs []int64) (*db.Channel, error) { + ch, err := s.DB.CreateGroupDMChannel(ctx, name, participantIDs) + if err == nil { + s.cancel() + } + return ch, err +} + +// OC-0004: CreateGroupDMChannel commits the channel, all dm_participants rows +// and all dm_open_state rows in one transaction. The subsequent +// GetDMParticipants read used to run on the same cancellable request context, +// so a client disconnect landing right after the commit (context cancelled +// in the gap) turned a fully-persisted group DM into a reported failure — +// inviting a client retry that, because group DMs are duplicate-by-design +// (db/dm_queries.go CreateGroupDMChannel doc), creates a second identical +// group. +func TestDMService_CreateGroupDM_SurvivesCancelledPostCommitRead(t *testing.T) { + database := newTestDB(t) + seedUser(t, database, &db.User{ID: 1, Username: "alice"}) + seedUser(t, database, &db.User{ID: 2, Username: "bob"}) + seedUser(t, database, &db.User{ID: 3, Username: "carol"}) + + ctx, cancel := context.WithCancel(context.Background()) + st := &cancelAfterCreateGroupDMStore{DB: database} + st.cancel = cancel + + svc := NewDMService(st) + result, err := svc.CreateGroupDM(ctx, 1, []int64{2, 3}, "") + if err != nil { + t.Fatalf("CreateGroupDM with context cancelled right after commit: %v (the channel is already persisted at this point — this must not fail the request)", err) + } + if result.Channel == nil { + t.Fatal("expected a created channel even though the post-commit context was cancelled") + } + if len(result.ParticipantIDs) != 3 { + t.Fatalf("ParticipantIDs = %v, want 3 entries so the caller can still broadcast dm_channel_open", result.ParticipantIDs) + } + + // The channel must actually be persisted — a retry after this "failure" + // would otherwise be indistinguishable from creating a brand new group. + var count int + if err := database.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM dm_participants WHERE channel_id = ?`, result.Channel.ID, + ).Scan(&count); err != nil { + t.Fatalf("count participants: %v", err) + } + if count != 3 { + t.Fatalf("persisted participant rows = %d, want 3", count) + } +} diff --git a/Server/ws/emit.go b/Server/ws/emit.go index b04dcf9b..efbd1fc3 100644 --- a/Server/ws/emit.go +++ b/Server/ws/emit.go @@ -32,12 +32,33 @@ func (h *Hub) EmitEvents(ctx context.Context, events []Event) { // An invisible user's public presence half rides this branch; // like the visible case below, it must invalidate any queued // coalescer entry so a stale connect-time presence can't flush - // after (and overwrite) this fresher user-chosen status. + // after (and overwrite) this fresher user-chosen status. The + // drop and the broadcast run atomically under presenceMu (see + // dropQueuedPresenceAndBroadcast, OC-0005) so a flush racing in + // at the same moment can never enqueue its stale frame after + // this fresher one. if po, isPresence := ev.(PresenceOthersEvent); isPresence { - h.dropQueuedPresence(po.excludeUserID) + h.dropQueuedPresenceAndBroadcast(po.excludeUserID, func() { + // Normal priority, excluding the owner — NOT + // broadcastExcludeLow. Every other source of this same + // user's presence (connect/disconnect via BroadcastToAll, + // and the visible presence_update path below) already + // shares the normal-priority queue; putting this one on + // the low-priority queue instead split one user's + // presence across two per-client FIFOs with different + // durability (low silently drops on overflow instead of + // disconnecting, so no replay ever repairs the loss) and + // different drain order (writePump always drains normal + // strictly before low, so a newer frame on one queue can + // be delivered before an older frame still sitting on the + // other) — exactly the hazard OC-0214 fixed for the + // visible case below (OC-0003). + h.BroadcastToAllExcept(po.excludeUserID, e.Payload()) + }) + } else { + // Low priority: typing indicators are ephemeral. + h.broadcastExcludeLow(e.ChannelID(), e.ExcludeUserID(), e.Payload()) } - // Low priority: typing indicators are ephemeral. - h.broadcastExcludeLow(e.ChannelID(), e.ExcludeUserID(), e.Payload()) case UserTargetedEvent: // High priority: targeted events (DM opens, mentions). // dm_channel_open is unsequenced and targeted, so replay can never @@ -75,12 +96,18 @@ func (h *Hub) EmitEvents(ctx context.Context, events []Event) { // single ordered, seq-stamped, replayable stream (OC-0214). if pe, isPresence := ev.(PresenceEvent); isPresence { // A user-chosen presence also bypasses the connect/disconnect - // coalescer; drop any entry still queued for this user or the - // pending flush (up to 300ms later) would overwrite this - // fresher status with the stale connect-time one. - h.dropQueuedPresence(pe.userID) + // coalescer; drop any entry still queued for this user and + // broadcast atomically under presenceMu (see + // dropQueuedPresenceAndBroadcast, OC-0005), or the pending + // flush (up to 300ms later) could race in between the drop + // and the broadcast and overwrite this fresher status with + // the stale connect-time one. + h.dropQueuedPresenceAndBroadcast(pe.userID, func() { + h.BroadcastToAll(e.Payload()) + }) + } else { + h.BroadcastToAll(e.Payload()) } - h.BroadcastToAll(e.Payload()) default: slog.Warn("EmitEvents: unknown event type", "type", fmt.Sprintf("%T", ev)) } diff --git a/Server/ws/emit_presence_others_priority_test.go b/Server/ws/emit_presence_others_priority_test.go new file mode 100644 index 00000000..8e0cac01 --- /dev/null +++ b/Server/ws/emit_presence_others_priority_test.go @@ -0,0 +1,96 @@ +package ws + +// emit_presence_others_priority_test.go — regression test for OC-0003: the +// public half of an invisible user's presence (PresenceOthersEvent) went out +// via h.broadcastExcludeLow onto the low-priority queue — the same ephemeral, +// drop-on-overflow, unsequenced transport used for typing indicators — while +// every other source of the same user's presence (connect/disconnect via +// BroadcastToAll, and the visible presence_update path OC-0214 already fixed) +// shares the normal-priority queue. That split one user's presence across two +// per-client FIFOs with different durability and different drain order: +// writePump always drains normal strictly before low, so an observer with an +// older frame on low and a newer one on normal (or vice versa) can end up +// with the wrong one landing last, and a full low-priority queue silently +// drops the frame with no seq bump and no replay recovery — unlike the +// normal-priority queue, which disconnects on overflow and lets replay (or a +// fresh ready) repair the gap. + +import ( + "context" + "testing" + "time" +) + +// TestEmitEvents_PresenceOthersEvent_UsesNormalPriorityQueue pins the fix: a +// PresenceOthersEvent (the public half of an invisible presence change), +// routed through the ExcludeSenderEvent case in EmitEvents, must land on an +// observer's normal-priority queue — the same FIFO every other presence +// source for that user uses — never the low-priority queue, while still never +// reaching the excluded user (the invisible owner) themselves. +// +// Before the fix, emit.go special-cased this case onto h.broadcastExcludeLow, +// so this test observes the frame on the observer's c.sendLow instead of +// c.send, and fails. +func TestEmitEvents_PresenceOthersEvent_UsesNormalPriorityQueue(t *testing.T) { + h := newEmitTestHub() + + // Built directly (not via the emit_test.go helpers) so send and sendLow + // are DISTINCT channels — the shared-channel helpers in export_test.go + // unify them "for test observability" and would mask exactly the + // queue-split this test needs to detect. + observer := &Client{ + hub: h, + ctx: context.Background(), + userID: 1, + send: make(chan []byte, 8), + sendHigh: make(chan []byte, 8), + sendLow: make(chan []byte, 8), + } + owner := &Client{ + hub: h, + ctx: context.Background(), + userID: 2, + send: make(chan []byte, 8), + sendHigh: make(chan []byte, 8), + sendLow: make(chan []byte, 8), + } + h.clients[1] = observer + h.clients[2] = owner + h.pubsub.Subscribe(observer, TopicGlobal) + h.pubsub.Subscribe(owner, TopicGlobal) + + // The normal-priority path goes through the async hub.broadcast channel, + // so the hub loop must be running to deliver it. + go h.Run() + defer h.Stop() + + payload := []byte(`{"type":"presence","user_id":2,"status":"offline"}`) + h.EmitEvents(context.Background(), []Event{ + PresenceOthersEvent{excludeUserID: 2, payload: payload}, + }) + + observerNormal := drainChan(observer.send, 200*time.Millisecond) + observerLow := drainChan(observer.sendLow, 50*time.Millisecond) + ownerNormal := drainChan(owner.send, 50*time.Millisecond) + ownerLow := drainChan(owner.sendLow, 50*time.Millisecond) + + if len(observerNormal) != 1 { + t.Errorf("expected the public half of an invisible presence change on the "+ + "observer's normal-priority queue (same FIFO as connect/disconnect "+ + "presence), got %d normal messages, %d low messages", + len(observerNormal), len(observerLow)) + } + if len(observerLow) != 0 { + t.Errorf("invisible presence's public half must not go out on the "+ + "low-priority queue: writePump drains normal strictly before low, so "+ + "a frame queued there can be delivered after a later connect/"+ + "disconnect presence frame on the normal queue, leaving the "+ + "observer's final view stale, and is silently dropped (no replay "+ + "recovery) on overflow; got %d low messages", len(observerLow)) + } + if len(ownerNormal) != 0 || len(ownerLow) != 0 { + t.Errorf("the excluded owner must never receive the public half of "+ + "their own invisible presence change: got %d normal, %d low messages", + len(ownerNormal), len(ownerLow)) + } +} diff --git a/Server/ws/event.go b/Server/ws/event.go index 604d62ce..8fe90857 100644 --- a/Server/ws/event.go +++ b/Server/ws/event.go @@ -247,8 +247,12 @@ func (e PresenceEvent) Payload() []byte { return e.payload } // PresenceOthersEvent is the public half of an invisible user's presence: the // mapped ("offline") payload, broadcast to everyone except the user it -// describes. Satisfies ExcludeSenderEvent with a channel id of 0, which -// broadcastExcludeLow routes as a global publish minus one subscriber. +// describes. Satisfies ExcludeSenderEvent with a channel id of 0; EmitEvents +// special-cases it onto h.BroadcastToAllExcept (normal priority, sequenced, +// replayable) rather than the ExcludeSenderEvent default of +// broadcastExcludeLow, so this frame shares the same durable per-client FIFO +// as every other source of the same user's presence instead of the +// ephemeral, drop-on-overflow one built for typing indicators (OC-0003). type PresenceOthersEvent struct { excludeUserID int64 payload []byte diff --git a/Server/ws/hub_broadcast.go b/Server/ws/hub_broadcast.go index d168defc..d0dcdee3 100644 --- a/Server/ws/hub_broadcast.go +++ b/Server/ws/hub_broadcast.go @@ -20,6 +20,13 @@ type broadcastMsg struct { // recipient's role may not READ, and the audience is resolved off the hub // goroutine so deliverBroadcast stays free of permission queries. recipients []int64 + // excludeUserID, when non-zero, is omitted from a global (channelID == 0) + // broadcast's live delivery. Used for the public half of an invisible + // user's presence (see BroadcastToAllExcept): everyone else must see it, + // but the owner's own view comes from a separate, synchronous, targeted + // send, and the two racing would let the async global broadcast overwrite + // it. Ignored outside the channelID == 0 branch of deliverBroadcast. + excludeUserID int64 // enqueuedAt stamps the enqueue site so deliverBroadcast can record // enqueue→fanout latency. Zero on test-constructed messages; skipped then. enqueuedAt time.Time @@ -50,6 +57,29 @@ func (h *Hub) BroadcastToAll(msg []byte) { } } +// BroadcastToAllExcept enqueues msg for delivery to every connected client +// except excludeUserID. Non-blocking, like BroadcastToAll: if the broadcast +// channel is full the message is dropped with a warning. +// +// Routes through the SAME h.broadcast channel — and so the same single- +// goroutine hub dispatch loop and seqMu-serialized deliverBroadcast — as +// BroadcastToAll and every other normal-priority global broadcast. That +// shared serialization is what gives two broadcasts about the same user (say, +// a connect/disconnect presence frame and this one) their correct relative +// order at each observer: whichever enqueues onto h.broadcast first is also +// delivered to c.send first. A caller that instead published straight to +// pub/sub, bypassing this queue, would reintroduce exactly that kind of +// reordering from the other direction (OC-0003). +func (h *Hub) BroadcastToAllExcept(excludeUserID int64, msg []byte) { + select { + case h.broadcast <- broadcastMsg{channelID: 0, excludeUserID: excludeUserID, msg: msg, enqueuedAt: time.Now()}: + default: + h.broadcastDrops.Add(1) + slog.Warn("hub: broadcast channel full, dropping global message", + "msg_len", len(msg)) + } +} + // broadcastVoiceEvent enqueues a voice_state / voice_leave message for the // connected clients whose current role may READ channelID. // @@ -599,27 +629,71 @@ func (h *Hub) QueuePresence(userID int64, status string, customStatus *string) { } } -// dropQueuedPresence removes a user's pending coalesced presence, if any. -// Called when a fresher presence for that user is broadcast directly (the -// presence_update handler path), so the coalescer's later flush cannot -// resurrect the stale connect-time state over it. Ordering holds because a -// user's connect (which queues) and their presence_update (which drops) run -// serially on the same connection's readPump. -func (h *Hub) dropQueuedPresence(userID int64) { +// dropQueuedPresenceAndBroadcast atomically removes any coalesced presence +// still queued for userID and runs broadcast, both under presenceMu. Called +// when a fresher presence for that user is delivered directly (the +// presence_update handler path, via EmitEvents), so the delete and the send +// of the fresher frame can never straddle flushPresenceQueue's own +// snapshot-and-broadcast critical section (OC-0005). +// +// Holding presenceMu across the delete AND the broadcast — rather than just +// the delete — is what actually closes the race: whichever of this call and +// flushPresenceQueue acquires presenceMu second also enqueues its broadcast +// second. +// - If this call goes first, it deletes the entry before flush can ever +// snapshot it, so flush never broadcasts the stale state at all. +// - If flush goes first, this call's delete is a no-op against the +// already-cleared queue, but its broadcast still cannot run until flush's +// own broadcast has already been enqueued — so the fresher frame is +// stamped with the higher seq by deliverBroadcast's single FIFO consumer +// and every client's final view converges on it, not the stale one. +// +// broadcast runs with presenceMu held: every current caller (BroadcastToAll, +// BroadcastToAllExcept) only enqueues onto h.broadcast's non-blocking +// channel send, so this cannot block and introduces no new lock-order edge. +// Both callers sharing that same channel also means the "enqueues second" +// ordering guarantee above translates directly into delivery order: both +// broadcasts are drained by the same single-consumer hub dispatch loop +// (deliverBroadcast), in the order they were enqueued. +func (h *Hub) dropQueuedPresenceAndBroadcast(userID int64, broadcast func()) { h.presenceMu.Lock() + defer h.presenceMu.Unlock() delete(h.presenceQueue, userID) - h.presenceMu.Unlock() + broadcast() } +// presenceFlushRaceHook, when non-nil, runs once per flushPresenceQueue call +// immediately after the coalesced queue has been snapshotted and cleared, +// while presenceMu is still held. Test-only (always nil in production): the +// snapshot-to-broadcast window is too narrow to land a real concurrent +// dropQueuedPresenceAndBroadcast reliably, so tests use this hook to +// reproduce that interleaving deterministically. Mirrors the established +// refreshChannelVisibilityRaceHook / voiceJoinPostTokenRaceHook pattern. +var presenceFlushRaceHook func() + // flushPresenceQueue drains the coalescer and broadcasts each user's latest -// presence. Runs on the AfterFunc timer goroutine, never under presenceMu -// during the fan-out. +// presence, all under presenceMu (OC-0005). Runs on the AfterFunc timer +// goroutine. +// +// presenceMu is held across the broadcast loop, not just the snapshot: it +// used to be released beforehand, which let a concurrent +// dropQueuedPresenceAndBroadcast (nee dropQueuedPresence) call race in after +// the snapshot had already escaped the lock. The drop was then a guaranteed +// no-op against the live (already-nilled) map, AND nothing constrained +// whether that call's own fresher broadcast landed on h.broadcast before or +// after this loop's stale one — so the stale connect-time presence could win +// the seq race and permanently overwrite a user-chosen status. Holding the +// lock here forces the two critical sections to serialize, which is what +// dropQueuedPresenceAndBroadcast's ordering guarantee depends on. func (h *Hub) flushPresenceQueue() { h.presenceMu.Lock() + defer h.presenceMu.Unlock() queued := h.presenceQueue h.presenceQueue = nil h.presenceFlushArmed = false - h.presenceMu.Unlock() + if presenceFlushRaceHook != nil { + presenceFlushRaceHook() + } for uid, p := range queued { h.BroadcastPresence(uid, p.status, p.customStatus) } @@ -643,7 +717,15 @@ func (h *Hub) BroadcastPresence(userID int64, status string, customStatus *strin // presencePayload.CustomStatus has no omitempty) so the client clears any // cached text, matching what db.MemberSummary.ForViewer already does for // the ready payload's member list. - h.broadcastExcludeLow(0, userID, buildPresenceMsg(userID, public, nil)) + // + // Normal priority, excluding the owner (BroadcastToAllExcept), not + // broadcastExcludeLow: the low-priority queue is unsequenced and dropped + // (not disconnected) on overflow, so it could silently lose this frame + // with no replay recovery, and — since writePump always drains normal + // strictly before low — deliver it out of order against the very + // connect/disconnect presence frames this same coalescer flush also + // produces for other users via BroadcastToAll (OC-0003). + h.BroadcastToAllExcept(userID, buildPresenceMsg(userID, public, nil)) h.SendToUser(userID, buildPresenceMsg(userID, status, customStatus)) } @@ -858,8 +940,12 @@ func (h *Hub) deliverBroadcast(bm broadcastMsg) { h.SendToUser(userID, msg) } case bm.channelID == 0: - // Global broadcast — deliver to every connected client. - h.pubsub.PublishGlobal(msg) + // Global broadcast — deliver to every connected client, minus + // excludeUserID when the caller set one (see BroadcastToAllExcept). + // Publish(TopicGlobal, msg, 0) is exactly PublishGlobal(msg) when + // excludeUserID is the zero value, so ordinary BroadcastToAll + // callers are unaffected. + h.pubsub.Publish(TopicGlobal, msg, bm.excludeUserID) default: // Channel-scoped broadcast — deliver to subscribers of the channel // topic. The rate limiter already passed above, before the seq diff --git a/Server/ws/oc_0006_video_stream_count_test.go b/Server/ws/oc_0006_video_stream_count_test.go new file mode 100644 index 00000000..cb5934a3 --- /dev/null +++ b/Server/ws/oc_0006_video_stream_count_test.go @@ -0,0 +1,119 @@ +package ws + +// OC-0006: EnableCameraIfUnderLimit / EnableScreenshareIfUnderLimit gate on +// COUNT(*) FROM voice_states WHERE ... (camera = 1 OR screenshare = 1) < N — +// one row per *user*, not one unit per *stream*. camera and screenshare are +// independent columns on the same row, so a single user with both flags set +// consumes only one slot in the count while actually publishing two streams. +// A channel capped at N simultaneous video streams can therefore over-admit +// to 2N live streams while still refusing the next publisher, claiming the +// N-stream cap is reached when more than N streams are already live. + +import ( + "context" + "testing" + + "github.com/owncord/server/db" + "github.com/owncord/server/permissions" +) + +const oc0006VideoStreamRoleID = int64(211) + +func newOC0006VideoStreamDB(t *testing.T) *db.DB { + t.Helper() + database, err := db.Open(":memory:") + if err != nil { + t.Fatalf("db.Open: %v", err) + } + if err := db.Migrate(database); err != nil { + t.Fatalf("Migrate: %v", err) + } + t.Cleanup(func() { _ = database.Close() }) + if _, err := database.ExecContext(context.Background(), + `INSERT INTO roles (id, name, color, permissions, position, is_default) + VALUES (?, 'oc-0006-video', NULL, ?, 5, 0)`, + oc0006VideoStreamRoleID, + permissions.ReadMessages|permissions.ConnectVoice|permissions.SpeakVoice|permissions.UseVideo|permissions.ShareScreen, + ); err != nil { + t.Fatalf("seed oc-0006-video role: %v", err) + } + return database +} + +func seedOC0006VideoStreamUser(t *testing.T, database *db.DB, username string) int64 { + t.Helper() + uid, err := database.CreateUser(context.Background(), username, "hash", int(oc0006VideoStreamRoleID)) + if err != nil { + t.Fatalf("CreateUser %s: %v", username, err) + } + return uid +} + +// mustCreateVideoCappedChannel2 mirrors mustCreateVideoCappedChannel from +// oc_0023_screenshare_video_limit_test.go (kept file-local to avoid a +// cross-file test helper dependency). +func mustCreateVideoCappedChannel2(t *testing.T, database *db.DB, name string, maxVideo int) int64 { + t.Helper() + chID, err := database.CreateChannel(context.Background(), name, "voice", "", "", 0) + if err != nil { + t.Fatalf("CreateChannel %s: %v", name, err) + } + if err := database.AdminUpdateChannel(context.Background(), chID, db.ChannelUpdate{ + Name: name, + VoiceMaxVideo: maxVideo, + }); err != nil { + t.Fatalf("AdminUpdateChannel %s: %v", name, err) + } + return chID +} + +// A channel capped at 2 simultaneous video streams. Alice alone publishes +// both her camera and her screenshare -- that is 2 streams from one row. +// Bob's camera enable must then be refused: the cap is already saturated by +// Alice's two streams. Today the gate counts Alice's row once (camera=1 OR +// screenshare=1 matches her single row), so it reads slot usage as 1, not 2, +// and wrongly admits Bob's third stream. +func TestEnableVideoSlot_SameUserDoubleStreamCountsTwoSlots(t *testing.T) { + ctx := context.Background() + database := newOC0006VideoStreamDB(t) + chID := mustCreateVideoCappedChannel2(t, database, "capped-room-samerow", 2) + + alice := seedOC0006VideoStreamUser(t, database, "alice-double") + bob := seedOC0006VideoStreamUser(t, database, "bob-third") + if err := database.JoinVoiceChannel(ctx, alice, chID); err != nil { + t.Fatalf("JoinVoiceChannel alice: %v", err) + } + if err := database.JoinVoiceChannel(ctx, bob, chID); err != nil { + t.Fatalf("JoinVoiceChannel bob: %v", err) + } + + d := VoiceDeps{DB: database, Permissions: permissions.NewChecker(database)} + + ssRes := handleVoiceScreenshareV2(ctx, VoiceScreenshareCmd{userID: alice, enabled: true}, ClientInfo{UserID: alice, VoiceChannelID: chID}, d) + if ssRes.Error != nil { + t.Fatalf("alice screenshare enable (1st stream, cap 2) should succeed, got error: %+v", ssRes.Error) + } + + camRes := handleVoiceCameraV2(ctx, VoiceCameraCmd{userID: alice, enabled: true}, ClientInfo{UserID: alice, VoiceChannelID: chID}, d) + if camRes.Error != nil { + t.Fatalf("alice camera enable (2nd stream, cap 2) should succeed, got error: %+v", camRes.Error) + } + + // Cap is now saturated: Alice alone is publishing 2 of the 2 allowed + // streams. Bob's camera enable is a 3rd stream and must be refused. + bobCamRes := handleVoiceCameraV2(ctx, VoiceCameraCmd{userID: bob, enabled: true}, ClientInfo{UserID: bob, VoiceChannelID: chID}, d) + if bobCamRes.Error == nil { + t.Fatal("bob's camera enable succeeded as the channel's 3rd live video stream against a cap of 2 -- the same-user double-publish (camera+screenshare on one row) was undercounted as a single slot") + } + if ce, ok := bobCamRes.Error.(ClientError); !ok || ce.Code != ErrCodeVideoLimit { + t.Errorf("error = %+v, want ClientError{Code: %q}", bobCamRes.Error, ErrCodeVideoLimit) + } + + vs, err := database.GetVoiceState(ctx, bob) + if err != nil || vs == nil { + t.Fatalf("GetVoiceState bob: %v", err) + } + if vs.Camera { + t.Error("bob's camera flag was set to true despite the VIDEO_LIMIT refusal") + } +} diff --git a/Server/ws/presence_coalesce_flush_race_test.go b/Server/ws/presence_coalesce_flush_race_test.go new file mode 100644 index 00000000..4c4a0b8e --- /dev/null +++ b/Server/ws/presence_coalesce_flush_race_test.go @@ -0,0 +1,97 @@ +package ws + +// presence_coalesce_flush_race_test.go — regression test for OC-0005. +// +// flushPresenceQueue snapshots h.presenceQueue and releases presenceMu +// BEFORE broadcasting the snapshotted entries. dropQueuedPresence (the guard +// EmitEvents uses to stop a stale connect/disconnect presence from +// clobbering a fresher user-chosen status) only deletes from the LIVE map, +// so a drop that lands after the flush has already taken its snapshot is a +// no-op — and, critically, nothing then constrains the relative order in +// which the flush's stale broadcast and the fresher direct broadcast reach +// h.broadcast. Both go through deliverBroadcast's single FIFO consumer, +// which stamps seq in enqueue order, so whichever one is enqueued LAST wins +// every client's final view. A stale connect-time presence enqueued after a +// user's own fresher presence_update therefore permanently overwrites it. +// +// The snapshot-to-broadcast window is a few instructions wide and not +// reliably landed by staggering real goroutines, so presenceFlushRaceHook +// (test-only, nil in production) fires at exactly that point, mirroring the +// established refreshChannelVisibilityRaceHook / voiceJoinPostTokenRaceHook +// pattern used to pin analogous races elsewhere in this package. + +import ( + "bytes" + "context" + "testing" + "time" +) + +// TestFlushPresenceQueue_ConcurrentDirectPresenceOrdersLast pins OC-0005 for +// the visible presence_update path (EmitEvents' BroadcastAllEvent branch). +// +// A stale connect-time "online" is queued for user 42. While +// flushPresenceQueue is mid-flush (queue already snapshotted), a concurrent +// presence_update to "dnd" races in via EmitEvents. The fresher "dnd" must +// end up enqueued on h.broadcast AFTER the stale "online" — so the hub +// stamps it with the higher seq and every other client's final view of user +// 42 converges on "dnd", not the stale "online". +func TestFlushPresenceQueue_ConcurrentDirectPresenceOrdersLast(t *testing.T) { + h := &Hub{ + broadcast: make(chan broadcastMsg, 8), + pubsub: NewPubSub(), + } + // Populate the queue directly rather than via QueuePresence: QueuePresence + // arms a real 300ms time.AfterFunc(h.flushPresenceQueue), and this test + // already drives flushPresenceQueue manually below. Leaving that timer + // armed would let it fire later — during a *later* run of this same test + // under -count=N, or during another test entirely — and invoke whatever + // presenceFlushRaceHook happens to be installed at that later moment, + // which is exactly the kind of cross-run interference this test must not + // introduce. + h.presenceMu.Lock() + h.presenceQueue = map[int64]pendingPresence{42: {status: "online"}} + h.presenceFlushArmed = true + h.presenceMu.Unlock() + + raced := make(chan struct{}) + var hookRan bool + presenceFlushRaceHook = func() { + hookRan = true + // Simulate EmitEvents' direct presence branch racing in exactly + // here: after flushPresenceQueue has snapshotted (and, currently, + // released presenceMu for) the queue, but before it has broadcast + // the stale entry. + go func() { + h.EmitEvents(context.Background(), presenceEvents(42, "dnd", nil)) + close(raced) + }() + // Give the goroutine room to actually run: with the bug this lets + // it complete its (unsynchronized) broadcast well before flush's + // own loop runs; with the fix in place the goroutine instead blocks + // on presenceMu until this function returns and releases it, so the + // sleep costs nothing extra there either way. + time.Sleep(20 * time.Millisecond) + } + defer func() { presenceFlushRaceHook = nil }() + + h.flushPresenceQueue() + <-raced + + if !hookRan { + t.Fatal("presenceFlushRaceHook never fired — test setup is broken, not exercising the flush race window") + } + + if len(h.broadcast) != 2 { + t.Fatalf("expected 2 broadcast frames (stale flush + fresh update), got %d", len(h.broadcast)) + } + first := <-h.broadcast + second := <-h.broadcast + + if !bytes.Contains(first.msg, []byte(`"status":"online"`)) { + t.Errorf("expected the stale flush's 'online' frame enqueued FIRST; got first=%s", first.msg) + } + if !bytes.Contains(second.msg, []byte(`"status":"dnd"`)) { + t.Errorf("expected the fresh presence_update's 'dnd' frame enqueued LAST (so it gets the higher seq and wins); got second=%s", second.msg) + } +} diff --git a/Server/ws/reconnect_fallback_channel_leak_test.go b/Server/ws/reconnect_fallback_channel_leak_test.go new file mode 100644 index 00000000..1e84f784 --- /dev/null +++ b/Server/ws/reconnect_fallback_channel_leak_test.go @@ -0,0 +1,184 @@ +package ws + +// reconnect_fallback_channel_leak_test.go — regression test for OC-0001. +// +// handleReconnect promotes an attacker-supplied (but READ-gated at the time) +// active_channel_id into c.channelID (serve.go ~349) before its own abort +// paths run. When a permission revocation lands deep in the handshake and +// trips the final mustFullResync re-check (serve.go ~401), handleReconnect +// aborts with handled=false and ServeWS falls through to handleFreshConnect +// — but nothing clears the already-set c.channelID. handleFreshConnect +// recomputes allowedChannelIDs WITHOUT the revoked channel and its +// buildReady payload correctly omits it, but registerNow (hub.go ~560) +// subscribes c.channelID's ChannelTopic unconditionally, with no +// readableChannelIDs check — unlike its two siblings in the same function. +// Every subsequent broadcast to that channel is then delivered to a client +// that was never granted READ_MESSAGES for it. + +import ( + "context" + "encoding/json" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/coder/websocket" + + "github.com/owncord/server/auth" + "github.com/owncord/server/permissions" +) + +func TestReconnect_FullReadyFallbackDoesNotLeakRevokedChannelSubscription(t *testing.T) { + database := newHarvestVoiceDB(t) + ctx := context.Background() + + uid := seedHarvestVoiceUser(t, database, "fallback-leak-user") + chID := mustCreateVoiceChannel(t, database, "fallback-leak-channel") + ch, err := database.GetChannel(ctx, chID) + if err != nil || ch == nil { + t.Fatalf("GetChannel: %v", err) + } + user, err := database.GetUserByID(ctx, uid) + if err != nil || user == nil { + t.Fatalf("GetUserByID: %v", err) + } + + token, err := auth.GenerateToken() + if err != nil { + t.Fatalf("GenerateToken: %v", err) + } + if _, err := database.CreateSession(ctx, uid, auth.HashToken(token), "test", "127.0.0.1"); err != nil { + t.Fatalf("CreateSession: %v", err) + } + + hub := NewHub(database, auth.NewRateLimiter(), nil) + go hub.Run() + defer hub.Stop() + + // Precondition: the channel is READ-visible at the moment the auth frame + // is evaluated, so the auth-frame active_channel_id is legitimately + // honoured by handleReconnect. + allowedBefore, err := hub.computeAllowedChannels(ctx, database, user) + if err != nil { + t.Fatalf("computeAllowedChannels: %v", err) + } + if !allowedBefore[chID] { + t.Fatalf("precondition: channel %d should start READ-visible", chID) + } + + // Bracket last_seq=99 so the resume takes the buffer tier (not a + // mustFullResync-forced full ready from the very start). + rb := hub.ReplayBuffer() + rb.Push(98, chID, []byte(`{"seq":98,"type":"chat_message","payload":{}}`)) + rb.Push(99, chID, []byte(`{"seq":99,"type":"chat_message","payload":{}}`)) + rb.Push(100, chID, []byte(`{"seq":100,"type":"chat_message","payload":{}}`)) + hub.SeedSeq(100) + const lastSeq = uint64(99) + + if hub.mustFullResync(lastSeq) { + t.Fatalf("precondition: mustFullResync must be false before any visibility change") + } + + // Fires once, deep inside handleReconnect — after c.channelID has already + // been promoted from active_channel_id (serve.go ~349) but before the + // final mustFullResync re-check (serve.go ~401). Revoke READ_MESSAGES on + // chID, mirroring an admin edit racing the resume, exactly like + // TestHandleReconnect_VisibilityChangeDuringHandshake_ForcesFullReady. + var hookRan bool + handleReconnectPreRegisterRaceHook = func() { + hookRan = true + if overrideErr := database.UpsertChannelOverride(ctx, chID, harvestVoiceRoleID, 0, permissions.ReadMessages); overrideErr != nil { + t.Fatalf("UpsertChannelOverride: %v", overrideErr) + } + //nolint:contextcheck // RefreshChannelVisibility takes no context by design. + hub.RefreshChannelVisibility(ch) + } + defer func() { handleReconnectPreRegisterRaceHook = nil }() + + srv := httptest.NewServer(ServeWS(hub, database, []string{"*"}, 0)) + defer srv.Close() + + dialCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + conn, dialResp, dialErr := websocket.Dial(dialCtx, "ws"+strings.TrimPrefix(srv.URL, "http"), nil) + if dialResp != nil && dialResp.Body != nil { + _ = dialResp.Body.Close() + } + if dialErr != nil { + t.Fatalf("websocket.Dial: %v", dialErr) + } + defer func() { _ = conn.Close(websocket.StatusNormalClosure, "") }() + + raw, _ := json.Marshal(map[string]any{ + "type": "auth", + "payload": map[string]any{ + "token": token, + "last_seq": lastSeq, + "active_channel_id": chID, + }, + }) + if err := conn.Write(dialCtx, websocket.MessageText, raw); err != nil { + t.Fatalf("write auth: %v", err) + } + + // The abort forces the fallback full-connect flow, which writes TWO + // frames — auth_ok, then ready — unlike the single auth_ok a successful + // buffer-tier resume would send. + readCtx, readCancel := context.WithTimeout(ctx, 5*time.Second) + defer readCancel() + _, authMsg, err := conn.Read(readCtx) + if err != nil { + t.Fatalf("read auth_ok: %v", err) + } + var authParsed map[string]any + if err := json.Unmarshal(authMsg, &authParsed); err != nil { + t.Fatalf("unmarshal auth_ok: %v", err) + } + if authParsed["type"] != MsgTypeAuthOK { + t.Fatalf("expected auth_ok, got %v", authParsed["type"]) + } + + _, readyMsg, err := conn.Read(readCtx) + if err != nil { + t.Fatalf("read ready: %v", err) + } + var readyParsed map[string]any + if err := json.Unmarshal(readyMsg, &readyParsed); err != nil { + t.Fatalf("unmarshal ready: %v", err) + } + if readyParsed["type"] != MsgTypeReady { + t.Fatalf("expected ready (handleReconnect aborted, fell through to handleFreshConnect), got %v", readyParsed["type"]) + } + + if !hookRan { + t.Fatal("handleReconnectPreRegisterRaceHook never fired — test setup is broken, not exercising the race window") + } + + deadline := time.Now().Add(2 * time.Second) + var c *Client + for { + hub.mu.Lock() + c = hub.clients[uid] + hub.mu.Unlock() + if c != nil { + break + } + if time.Now().After(deadline) { + t.Fatal("client was never registered") + } + time.Sleep(10 * time.Millisecond) + } + + if got := c.getChannelID(); got == chID { + t.Errorf("resumed-then-fallback client kept focus on revoked channel %d after registerNow, want 0", got) + } + + hub.pubsub.mu.RLock() + sub := hub.pubsub.topics[ChannelTopic(chID)][uid] + hub.pubsub.mu.RUnlock() + if sub != nil { + t.Errorf("client is subscribed to ChannelTopic(%d) despite READ_MESSAGES being revoked before registration — "+ + "every subsequent broadcast to that channel will be delivered to this socket", chID) + } +} diff --git a/Server/ws/serve.go b/Server/ws/serve.go index 9f139f2a..b57e53e6 100644 --- a/Server/ws/serve.go +++ b/Server/ws/serve.go @@ -692,6 +692,20 @@ func (h *Hub) handleFreshConnect( allowedChannelIDs = allowed } } + // handleReconnect may have promoted an auth-frame active_channel_id into + // c.channelID (serve.go, honoured only when it was READ-visible at that + // moment) and then aborted on one of its own re-checks — most notably the + // final mustFullResync check, tripped by a permission revocation that + // landed mid-handshake. None of those abort paths undo the c.channelID + // write. registerNow subscribes c.channelID's ChannelTopic + // unconditionally, so re-gate it here against the freshly recomputed + // permission set before registering. Fail closed: a nil allowedChannelIDs + // (lastSeq == 0, or the computeAllowedChannels error branch above) denies. + if chID := c.getChannelID(); chID != 0 && !allowedChannelIDs[chID] { + c.mu.Lock() + c.channelID = 0 + c.mu.Unlock() + } h.registerNow(c, allowedChannelIDs) // Settle the session's status before buildReady reads the member list, so diff --git a/Server/ws/serve_ready.go b/Server/ws/serve_ready.go index 7cab1316..82bc25a3 100644 --- a/Server/ws/serve_ready.go +++ b/Server/ws/serve_ready.go @@ -75,6 +75,29 @@ func (h *Hub) presentableMembers(members []db.MemberSummary, viewerID int64) []d return out } +// presentableDMChannels applies presentableMembers' "no live connection means +// offline" rule to a DM channel list's recipient statuses. GetUserDMChannels +// already applies db.StatusForViewer (the invisible-to-others half); this +// adds the missing "no live connection" half so dm_channels cannot disagree +// with the members array about whether the same disconnected user is online. +// Both Recipient (the legacy single-recipient field) and every entry of +// Recipients (the group-aware field) are rewritten, since a 1:1 DM's +// Recipient is a copy of Recipients[0], not a shared reference. +func (h *Hub) presentableDMChannels(dmChannels []db.DMChannelInfo) []db.DMChannelInfo { + connected := h.connectedUserIDs() + for i := range dmChannels { + if dmChannels[i].Recipient.ID != 0 && !connected[dmChannels[i].Recipient.ID] { + dmChannels[i].Recipient.Status = db.StatusOffline + } + for j := range dmChannels[i].Recipients { + if !connected[dmChannels[i].Recipients[j].ID] { + dmChannels[i].Recipients[j].Status = db.StatusOffline + } + } + } + return dmChannels +} + // connectedUserIDs snapshots the ids with a live WebSocket connection. func (h *Hub) connectedUserIDs() map[int64]bool { h.mu.RLock() @@ -249,6 +272,15 @@ func (h *Hub) buildReady(ctx context.Context, database *db.DB, userID int64, rol dmChannels[i].MentionCount = u.MentionCount } } + // GetUserDMChannels only applies db.StatusForViewer, which collapses + // invisible to offline but passes a disconnected recipient's saved + // idle/dnd through verbatim (MarkUserDisconnected deliberately keeps a + // chosen idle/dnd across a disconnect so the next connect can honour it, + // relying on every read path to hide it in the meantime). members already + // gets the "no live connection means offline" half of that rule from + // presentableMembers above; apply the same half here so dm_channels + // cannot disagree with members about the same user within one payload. + dmChannels = h.presentableDMChannels(dmChannels) // Collect voice states, filtered to visible channels (BUG-095) plus the // user's own open DM channels — mirroring computeAllowedChannels, which diff --git a/Server/ws/serve_ready_dm_status_test.go b/Server/ws/serve_ready_dm_status_test.go new file mode 100644 index 00000000..22bffef9 --- /dev/null +++ b/Server/ws/serve_ready_dm_status_test.go @@ -0,0 +1,102 @@ +package ws_test + +// serve_ready_dm_status_test.go — regression test for OC-0008: buildReady's +// dm_channels half came straight from database.GetUserDMChannels, which +// applies only db.StatusForViewer — that collapses invisible to offline but +// passes a disconnected user's saved idle/dnd through verbatim. members goes +// through presentableMembers first, which additionally forces offline for +// anyone with no live WebSocket connection (mirroring +// TestReady_DisconnectedMemberWithChosenStatusRendersOffline in +// presence_invisible_test.go, but for the dm_channels field). Before the fix +// a signed-out user who last chose "dnd" or "idle" would show as offline in +// members but still dnd/idle in dm_channels within the very same ready frame. + +import ( + "context" + "encoding/json" + "testing" + + "github.com/owncord/server/db" +) + +// dmChannelStatusFor pulls the recipient status for other.ID out of a ready +// payload's dm_channels array, checking both the legacy `recipient` field and +// the group-aware `recipients` array so a fix that only patches one leaks +// through undetected. +func dmChannelStatusFor(t *testing.T, raw []byte, otherID int64) (recipientStatus string, recipientsStatus string, found bool) { + t.Helper() + var env struct { + Payload struct { + DMChannels []struct { + Recipient struct { + ID int64 `json:"id"` + Status string `json:"status"` + } `json:"recipient"` + Recipients []struct { + ID int64 `json:"id"` + Status string `json:"status"` + } `json:"recipients"` + } `json:"dm_channels"` + } `json:"payload"` + } + if err := json.Unmarshal(raw, &env); err != nil { + t.Fatalf("unmarshal ready: %v", err) + } + for _, dm := range env.Payload.DMChannels { + if dm.Recipient.ID == otherID { + recipientStatus = dm.Recipient.Status + found = true + } + for _, r := range dm.Recipients { + if r.ID == otherID { + recipientsStatus = r.Status + } + } + } + return recipientStatus, recipientsStatus, found +} + +// TestBuildReady_DMChannelsHidesDisconnectedRecipientStatus pins OC-0008: a DM +// recipient with no live WebSocket connection must render offline in +// dm_channels, exactly as presentableMembers already forces for the members +// array. absent chooses "dnd", then MarkUserDisconnected-equivalent state is +// simulated by simply never registering a client for absent (buildReady's +// connectedUserIDs() only reflects live hub registrations, so an +// unregistered user is indistinguishable from "signed out"). +func TestBuildReady_DMChannelsHidesDisconnectedRecipientStatus(t *testing.T) { + hub, database := newServeHub(t) + ctx := context.Background() + + viewer := seedServeUser(t, database, "dm-status-viewer") + absent := seedServeUser(t, database, "dm-status-absent") + viewerRole, err := database.GetRoleByID(ctx, viewer.RoleID) + if err != nil || viewerRole == nil { + t.Fatalf("GetRoleByID: %v", err) + } + + // absent chose "dnd" before signing out; the column keeps it (this is + // exactly what MarkUserDisconnected leaves behind for a non-online status). + if err := database.UpdateUserStatus(ctx, absent.ID, db.StatusDND); err != nil { + t.Fatalf("UpdateUserStatus: %v", err) + } + + seedDMChannel(t, database, viewer.ID, absent.ID) + + // Only the viewer has a live connection; absent is never registered, so + // they must render offline everywhere in this ready payload. + msg, err := hub.BuildReadyWithRoleForTest(database, viewer.ID, viewerRole) + if err != nil { + t.Fatalf("BuildReadyWithRoleForTest: %v", err) + } + + recipientStatus, recipientsStatus, found := dmChannelStatusFor(t, msg, absent.ID) + if !found { + t.Fatalf("ready payload's dm_channels is missing recipient %d", absent.ID) + } + if recipientStatus != db.StatusOffline { + t.Errorf("dm_channels[].recipient.status = %q, want %q (disconnected member must render offline, same rule presentableMembers applies to the members array)", recipientStatus, db.StatusOffline) + } + if recipientsStatus != db.StatusOffline { + t.Errorf("dm_channels[].recipients[].status = %q, want %q", recipientsStatus, db.StatusOffline) + } +}