From deda095db148ad435173d69ec94bfa1814da4ae4 Mon Sep 17 00:00:00 2001 From: jevb Date: Tue, 31 Mar 2026 18:07:23 +0200 Subject: [PATCH] test: add WS coverage tests + refactor handlers, update docs Add hub, livekit, export, and coverage boost tests for Server/ws. Refactor handlers_chat.go and serve.go for testability. Sync docs: fix backup endpoint path, add DELETE /auth/account, add audit logging + account deletion to security.md. --- .gitignore | 3 + Server/ws/coverage_boost_test.go | 474 +++++++++++++++++++++++++++++++ Server/ws/export_test.go | 80 ++++++ Server/ws/handlers_chat.go | 251 ++++++++-------- Server/ws/hub_test.go | 123 ++++++++ Server/ws/livekit_test.go | 354 +++++++++++++++++++++++ Server/ws/serve.go | 244 +++++++--------- docs/api.md | 30 ++ docs/contributing.md | 3 +- docs/deployment.md | 10 +- docs/security.md | 15 + 11 files changed, 1324 insertions(+), 263 deletions(-) diff --git a/.gitignore b/.gitignore index b96e6c10..eca69f66 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,9 @@ CLAUDE.md .github/copilot-instructions.md .github/instructions/ +# Agent worktrees (Mission Control fleet isolation) +.worktrees/ + # AI-specific / internal planning docs docs/brain/ docs/CODEMAPS/ diff --git a/Server/ws/coverage_boost_test.go b/Server/ws/coverage_boost_test.go index dde922e8..04f1ba16 100644 --- a/Server/ws/coverage_boost_test.go +++ b/Server/ws/coverage_boost_test.go @@ -2220,3 +2220,477 @@ func TestHandleVoiceJoin_InvalidQualityFallsBackToMedium(t *testing.T) { } t.Error("expected voice_config with medium quality fallback") } + +// ─── getLastActivity (client.go:153) ───────────────────────────────────────── + +func TestGetLastActivity_ReturnsZeroForNewTestClient(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, 1, send) + + got := ws.GetLastActivityForTest(c) + if !got.IsZero() { + t.Fatalf("expected zero time for new test client, got %v", got) + } +} + +func TestGetLastActivity_UpdatedByTouch(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, 1, send) + + before := time.Now() + ws.TouchForTest(c) + after := time.Now() + + got := ws.GetLastActivityForTest(c) + if got.Before(before) || got.After(after) { + t.Fatalf("lastActivity = %v, expected between %v and %v", got, before, after) + } +} + +func TestGetLastActivity_MultipleTouch(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, 1, send) + + ws.TouchForTest(c) + first := ws.GetLastActivityForTest(c) + + time.Sleep(5 * time.Millisecond) + ws.TouchForTest(c) + second := ws.GetLastActivityForTest(c) + + if !second.After(first) { + t.Fatalf("second touch (%v) should be after first (%v)", second, first) + } +} + +// ─── setVoiceChID (client.go:186) ─────────────────────────────────────────── + +func TestSetVoiceChID_SetsAndGetsValue(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, 1, send) + + ws.SetVoiceChIDForTest(c, 77) + if got := ws.GetClientVoiceChIDForTest(c); got != 77 { + t.Fatalf("voiceChID = %d, want 77", got) + } +} + +func TestSetVoiceChID_OverwritesPreviousValue(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, 1, send) + + ws.SetVoiceChIDForTest(c, 10) + ws.SetVoiceChIDForTest(c, 20) + if got := ws.GetClientVoiceChIDForTest(c); got != 20 { + t.Fatalf("voiceChID = %d, want 20", got) + } +} + +func TestSetVoiceChID_ZeroMeansNotInVoice(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, 1, send) + + ws.SetVoiceChIDForTest(c, 50) + ws.SetVoiceChIDForTest(c, 0) + if got := ws.GetClientVoiceChIDForTest(c); got != 0 { + t.Fatalf("voiceChID = %d, want 0", got) + } +} + +// ─── clearVoiceChID (client.go:203) ───────────────────────────────────────── + +func TestClearVoiceChID_ReturnsOldValueAndClearsToZero(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, 1, send) + + ws.SetVoiceChIDForTest(c, 42) + old := ws.ClearVoiceChIDForTest(c) + if old != 42 { + t.Fatalf("clearVoiceChID returned %d, want 42", old) + } + if got := ws.GetClientVoiceChIDForTest(c); got != 0 { + t.Fatalf("voiceChID after clear = %d, want 0", got) + } +} + +func TestClearVoiceChID_ReturnsZeroWhenNotInVoice(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, 1, send) + + old := ws.ClearVoiceChIDForTest(c) + if old != 0 { + t.Fatalf("clearVoiceChID returned %d, want 0", old) + } +} + +func TestClearVoiceChID_DoubleClearReturnsZero(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, 1, send) + + ws.SetVoiceChIDForTest(c, 99) + first := ws.ClearVoiceChIDForTest(c) + second := ws.ClearVoiceChIDForTest(c) + if first != 99 { + t.Fatalf("first clear = %d, want 99", first) + } + if second != 0 { + t.Fatalf("second clear = %d, want 0", second) + } +} + +// ─── handleVoiceTokenRefresh (voice_join.go:188) ──────────────────────────── + +func voiceTokenRefreshMsg() []byte { + raw, _ := json.Marshal(map[string]any{ + "type": "voice_token_refresh", + "payload": map[string]any{}, + }) + return raw +} + +func TestHandleVoiceTokenRefresh_NotInVoice(t *testing.T) { + hub, database := newCoverageHub(t) + user := seedCoverageOwner(t, database, "vtr-notinvoice") + send := make(chan []byte, 16) + c := ws.NewTestClientWithUser(hub, user, 0, send) + hub.Register(c) + time.Sleep(20 * time.Millisecond) + + hub.HandleMessageForTest(c, voiceTokenRefreshMsg()) + time.Sleep(50 * time.Millisecond) + + code := drainForErrorCode(send, 200*time.Millisecond) + if code != "BAD_REQUEST" { + t.Errorf("error code = %q, want BAD_REQUEST", code) + } +} + +func TestHandleVoiceTokenRefresh_InVoice_ReturnsToken(t *testing.T) { + hub, database := newCoverageHub(t) + user := seedCoverageOwner(t, database, "vtr-invc") + vcID := seedVoiceChannel(t, database, "vtr-vc") + send := make(chan []byte, 64) + c := ws.NewTestClientWithUser(hub, user, 0, send) + hub.Register(c) + time.Sleep(20 * time.Millisecond) + + raw, _ := json.Marshal(map[string]any{ + "type": "voice_join", + "payload": map[string]any{"channel_id": vcID}, + }) + hub.HandleMessageForTest(c, raw) + time.Sleep(100 * time.Millisecond) + drainChanBuf(send) + + hub.HandleMessageForTest(c, voiceTokenRefreshMsg()) + time.Sleep(100 * time.Millisecond) + + msgs := drainChanTimeout(send, 300*time.Millisecond) + foundToken := false + for _, msg := range msgs { + var env map[string]any + if json.Unmarshal(msg, &env) == nil && env["type"] == "voice_token" { + foundToken = true + break + } + } + if !foundToken { + t.Error("expected voice_token message after token refresh") + } +} + +func TestHandleVoiceTokenRefresh_NilUser(t *testing.T) { + hub, _ := newCoverageHub(t) + send := make(chan []byte, 16) + c := ws.NewTestClient(hub, 1, send) + hub.Register(c) + time.Sleep(20 * time.Millisecond) + + ws.SetVoiceChIDForTest(c, 42) + + hub.HandleMessageForTest(c, voiceTokenRefreshMsg()) + time.Sleep(50 * time.Millisecond) + + code := drainForErrorCode(send, 200*time.Millisecond) + if code != "INTERNAL" { + t.Errorf("error code = %q, want INTERNAL", code) + } +} + +// ─── rollbackVoiceJoin (voice_join.go:239) ────────────────────────────────── + +func TestRollbackVoiceJoin_ClearsVoiceStateAndBroadcasts(t *testing.T) { + hub, database := newCoverageHub(t) + user := seedCoverageOwner(t, database, "rb-user") + vcID := seedVoiceChannel(t, database, "rb-vc") + send := make(chan []byte, 64) + c := ws.NewTestClientWithUser(hub, user, 0, send) + hub.Register(c) + time.Sleep(20 * time.Millisecond) + + if err := database.JoinVoiceChannel(user.ID, vcID); err != nil { + t.Fatalf("JoinVoiceChannel: %v", err) + } + ws.SetVoiceChIDForTest(c, vcID) + + hub.RollbackVoiceJoinForTest(c, vcID) + time.Sleep(100 * time.Millisecond) + + if got := ws.GetClientVoiceChIDForTest(c); got != 0 { + t.Fatalf("voiceChID after rollback = %d, want 0", got) + } + + state, _ := database.GetVoiceState(user.ID) + if state != nil { + t.Fatal("voice state should be nil after rollback") + } + + msgs := drainChanTimeout(send, 300*time.Millisecond) + foundLeave := false + for _, msg := range msgs { + var env map[string]any + if json.Unmarshal(msg, &env) == nil && env["type"] == "voice_leave" { + foundLeave = true + break + } + } + if !foundLeave { + t.Error("expected voice_leave broadcast after rollback") + } +} + +func TestRollbackVoiceJoin_NoDBState_DoesNotPanic(t *testing.T) { + hub, database := newCoverageHub(t) + user := seedCoverageOwner(t, database, "rb-nostate") + send := make(chan []byte, 16) + c := ws.NewTestClientWithUser(hub, user, 0, send) + hub.Register(c) + time.Sleep(20 * time.Millisecond) + + ws.SetVoiceChIDForTest(c, 999) + hub.RollbackVoiceJoinForTest(c, 999) + time.Sleep(50 * time.Millisecond) + + if got := ws.GetClientVoiceChIDForTest(c); got != 0 { + t.Fatalf("voiceChID after rollback = %d, want 0", got) + } +} + +// ─── leaveVoiceChannelWithRetry (voice_leave.go:57) ───────────────────────── + +func TestLeaveVoiceChannelWithRetry_SuccessOnFirstAttempt(t *testing.T) { + hub, database := newCoverageHub(t) + user := seedCoverageOwner(t, database, "lvcr-ok") + vcID := seedVoiceChannel(t, database, "lvcr-ok-vc") + + if err := database.JoinVoiceChannel(user.ID, vcID); err != nil { + t.Fatalf("JoinVoiceChannel: %v", err) + } + + state, _ := database.GetVoiceState(user.ID) + if state == nil { + t.Fatal("voice state should exist before leave") + } + + err := ws.LeaveVoiceChannelWithRetryForTest(hub, user.ID, vcID, state.JoinedAt) + if err != nil { + t.Fatalf("leaveVoiceChannelWithRetry returned error: %v", err) + } + + state, _ = database.GetVoiceState(user.ID) + if state != nil { + t.Fatal("voice state should be nil after successful leave") + } +} + +func TestLeaveVoiceChannelWithRetry_NoVoiceState_NilReturn(t *testing.T) { + hub, database := newCoverageHub(t) + _ = seedCoverageOwner(t, database, "lvcr-nostate") + + err := ws.LeaveVoiceChannelWithRetryForTest(hub, 9999, 1, "") + if err != nil { + t.Fatalf("expected nil error for non-existent voice state, got: %v", err) + } +} + +// ─── CleanupVoiceForChannel (hub.go:237) — additional paths ───────────────── + +func TestCleanupVoiceForChannel_WithClientsInChannel(t *testing.T) { + hub, database := newCoverageHub(t) + user1 := seedCoverageOwner(t, database, "cvfc-u1") + user2 := seedCoverageOwner(t, database, "cvfc-u2") + vcID := seedVoiceChannel(t, database, "cvfc-vc") + + send1 := make(chan []byte, 64) + send2 := make(chan []byte, 64) + c1 := ws.NewTestClientWithUser(hub, user1, 0, send1) + c2 := ws.NewTestClientWithUser(hub, user2, 0, send2) + hub.Register(c1) + hub.Register(c2) + time.Sleep(20 * time.Millisecond) + + if err := database.JoinVoiceChannel(user1.ID, vcID); err != nil { + t.Fatalf("JoinVoiceChannel u1: %v", err) + } + if err := database.JoinVoiceChannel(user2.ID, vcID); err != nil { + t.Fatalf("JoinVoiceChannel u2: %v", err) + } + ws.SetVoiceChIDForTest(c1, vcID) + ws.SetVoiceChIDForTest(c2, vcID) + + hub.CleanupVoiceForChannel(vcID) + time.Sleep(100 * time.Millisecond) + + if got := ws.GetClientVoiceChIDForTest(c1); got != 0 { + t.Errorf("c1 voiceChID = %d, want 0", got) + } + if got := ws.GetClientVoiceChIDForTest(c2); got != 0 { + t.Errorf("c2 voiceChID = %d, want 0", got) + } + + states, _ := database.GetChannelVoiceStates(vcID) + if len(states) != 0 { + t.Errorf("expected 0 voice states after cleanup, got %d", len(states)) + } +} + +func TestCleanupVoiceForChannel_EmptyChannel(t *testing.T) { + hub, database := newCoverageHub(t) + vcID := seedVoiceChannel(t, database, "cvfc-empty-vc") + hub.CleanupVoiceForChannel(vcID) + time.Sleep(20 * time.Millisecond) +} + +func TestCleanupVoiceForChannel_DBStateButNoClient(t *testing.T) { + hub, database := newCoverageHub(t) + user := seedCoverageOwner(t, database, "cvfc-noclient") + vcID := seedVoiceChannel(t, database, "cvfc-noclient-vc") + + if err := database.JoinVoiceChannel(user.ID, vcID); err != nil { + t.Fatalf("JoinVoiceChannel: %v", err) + } + + hub.CleanupVoiceForChannel(vcID) + time.Sleep(50 * time.Millisecond) + + state, _ := database.GetVoiceState(user.ID) + if state != nil { + t.Error("voice state should be nil after cleanup") + } +} + +// ─── sweepStaleVoiceStates (hub.go:489) ───────────────────────────────────── + +func TestSweepStaleVoiceStates_RemovesGhostState(t *testing.T) { + hub, database := newCoverageHub(t) + user := seedCoverageOwner(t, database, "sweep-ghost") + vcID := seedVoiceChannel(t, database, "sweep-ghost-vc") + + // Put user in voice in DB but don't register a client — ghost state. + if err := database.JoinVoiceChannel(user.ID, vcID); err != nil { + t.Fatalf("JoinVoiceChannel: %v", err) + } + + // Verify it exists. + state, _ := database.GetVoiceState(user.ID) + if state == nil { + t.Fatal("voice state should exist before sweep") + } + + hub.SweepStaleVoiceStatesForTest() + time.Sleep(100 * time.Millisecond) + + // Ghost state should be removed. + state, _ = database.GetVoiceState(user.ID) + if state != nil { + t.Error("ghost voice state should be nil after sweep") + } +} + +func TestSweepStaleVoiceStates_PreservesActiveClientState(t *testing.T) { + hub, database := newCoverageHub(t) + user := seedCoverageOwner(t, database, "sweep-active") + vcID := seedVoiceChannel(t, database, "sweep-active-vc") + + // Register client and set voice channel. + send := make(chan []byte, 64) + c := ws.NewTestClientWithUser(hub, user, 0, send) + hub.Register(c) + time.Sleep(20 * time.Millisecond) + + if err := database.JoinVoiceChannel(user.ID, vcID); err != nil { + t.Fatalf("JoinVoiceChannel: %v", err) + } + ws.SetVoiceChIDForTest(c, vcID) + + hub.SweepStaleVoiceStatesForTest() + time.Sleep(100 * time.Millisecond) + + // Active client's state should be preserved. + state, _ := database.GetVoiceState(user.ID) + if state == nil { + t.Error("active client's voice state should be preserved after sweep") + } +} + +func TestSweepStaleVoiceStates_NoStatesNoPanic(t *testing.T) { + hub, _ := newCoverageHub(t) + hub.SweepStaleVoiceStatesForTest() +} + +func TestSweepStaleVoiceStates_MismatchedChannelIsGhost(t *testing.T) { + hub, database := newCoverageHub(t) + user := seedCoverageOwner(t, database, "sweep-mismatch") + vc1 := seedVoiceChannel(t, database, "sweep-mismatch-vc1") + vc2 := seedVoiceChannel(t, database, "sweep-mismatch-vc2") + + // Register client in vc1 but DB says vc2. + send := make(chan []byte, 64) + c := ws.NewTestClientWithUser(hub, user, 0, send) + hub.Register(c) + time.Sleep(20 * time.Millisecond) + + if err := database.JoinVoiceChannel(user.ID, vc2); err != nil { + t.Fatalf("JoinVoiceChannel: %v", err) + } + ws.SetVoiceChIDForTest(c, vc1) // Client thinks vc1, DB says vc2 — mismatch. + + hub.SweepStaleVoiceStatesForTest() + time.Sleep(100 * time.Millisecond) + + // Mismatched state should be removed from DB. + state, _ := database.GetVoiceState(user.ID) + if state != nil { + t.Error("mismatched voice state should be removed after sweep") + } +} + +// ─── BroadcastToChannel / BroadcastToAll full-channel path ────────────────── + +func TestBroadcastToChannel_DropsWhenFull(t *testing.T) { + hub, _ := newCoverageHub(t) + // Don't start Run() — broadcast channel will fill up. + // The broadcast channel capacity is 256. + for i := 0; i < 260; i++ { + hub.BroadcastToChannel(1, []byte(`{"type":"test"}`)) + } + // No panic and no block = pass. Some messages will be dropped. +} + +func TestBroadcastToAll_DropsWhenFull(t *testing.T) { + hub, _ := newCoverageHub(t) + for i := 0; i < 260; i++ { + hub.BroadcastToAll([]byte(`{"type":"test"}`)) + } +} + + diff --git a/Server/ws/export_test.go b/Server/ws/export_test.go index ea06c3a5..bfe4d9a2 100644 --- a/Server/ws/export_test.go +++ b/Server/ws/export_test.go @@ -5,12 +5,92 @@ package ws import ( "encoding/json" "fmt" + "os/exec" "time" "github.com/livekit/protocol/livekit" "github.com/owncord/server/db" ) +// ─── hub sweep helpers ───────────────────────────────────────────────────── + +// SweepStaleClientsForTest exposes sweepStaleClients for external tests. +func (h *Hub) SweepStaleClientsForTest() { + h.sweepStaleClients() +} + +// SweepStaleVoiceStatesForTest exposes sweepStaleVoiceStates for external tests. +func (h *Hub) SweepStaleVoiceStatesForTest() { + h.sweepStaleVoiceStates() +} + +// SetClientLastActivityForTest overwrites a client's lastActivity timestamp. +func SetClientLastActivityForTest(c *Client, t time.Time) { + c.mu.Lock() + defer c.mu.Unlock() + c.lastActivity = t +} + +// ─── client getter/setter helpers ────────────────────────────────────────── + +// GetLastActivityForTest exposes Client.getLastActivity for external tests. +func GetLastActivityForTest(c *Client) time.Time { + return c.getLastActivity() +} + +// ClearVoiceChIDForTest exposes Client.clearVoiceChID for external tests. +func ClearVoiceChIDForTest(c *Client) int64 { + return c.clearVoiceChID() +} + +// SetVoiceChIDForTest exposes Client.setVoiceChID for external tests. +func SetVoiceChIDForTest(c *Client, chID int64) { + c.setVoiceChID(chID) +} + +// TouchForTest exposes Client.touch for external tests. +func TouchForTest(c *Client) { + c.touch() +} + +// RollbackVoiceJoinForTest exposes Hub.rollbackVoiceJoin for external tests. +func (h *Hub) RollbackVoiceJoinForTest(c *Client, channelID int64) { + h.rollbackVoiceJoin(c, channelID) +} + +// LeaveVoiceChannelWithRetryForTest exposes leaveVoiceChannelWithRetry for external tests. +func LeaveVoiceChannelWithRetryForTest(h *Hub, userID int64, channelID int64, joinToken string) error { + return leaveVoiceChannelWithRetry(h, userID, channelID, joinToken) +} + +// ─── livekit process/webhook helpers ─────────────────────────────────────── + +// GenerateConfigForTest exposes LiveKitProcess.generateConfig for external tests. +func (p *LiveKitProcess) GenerateConfigForTest() (string, error) { + return p.generateConfig() +} + +// SetProcessCmdForTest sets cmd to a non-nil value to simulate "already running". +func (p *LiveKitProcess) SetProcessCmdForTest() { + p.mu.Lock() + defer p.mu.Unlock() + p.cmd = &exec.Cmd{} +} + +// SetProcessStoppedForTest sets stopped=true to simulate a stopped process. +func (p *LiveKitProcess) SetProcessStoppedForTest() { + p.mu.Lock() + defer p.mu.Unlock() + p.stopped = true +} + +// NewHubForTest creates a minimal Hub with no DB or limiter for webhook testing. +func NewHubForTest() *Hub { + return &Hub{ + clients: make(map[int64]*Client), + } +} + // BuildAuthOKForTest exposes Hub.buildAuthOK for external tests. func (h *Hub) BuildAuthOKForTest(user *db.User, roleName string) []byte { return h.buildAuthOK(user, roleName) diff --git a/Server/ws/handlers_chat.go b/Server/ws/handlers_chat.go index 271039e8..15774c4e 100644 --- a/Server/ws/handlers_chat.go +++ b/Server/ws/handlers_chat.go @@ -7,6 +7,7 @@ import ( "log/slog" "time" + "github.com/owncord/server/db" "github.com/owncord/server/permissions" ) @@ -23,21 +24,22 @@ func registerChatHandlers(r *HandlerRegistry) { }) } +type chatSendPayload struct { + ChannelID json.Number `json:"channel_id"` + Content string `json:"content"` + ReplyTo *int64 `json:"reply_to"` + Attachments []string `json:"attachments"` +} + // handleChatSend processes a chat_send message. func (h *Hub) handleChatSend(ctx context.Context, c *Client, reqID string, payload json.RawMessage) { - // Rate limit. ratKey := fmt.Sprintf("chat:%d", c.userID) if !h.limiter.Allow(ratKey, chatRateLimit, chatWindow) { c.sendMsg(buildRateLimitError("too many messages", chatWindow.Seconds())) return } - var p struct { - ChannelID json.Number `json:"channel_id"` - Content string `json:"content"` - ReplyTo *int64 `json:"reply_to"` - Attachments []string `json:"attachments"` - } + var p chatSendPayload if err := json.Unmarshal(payload, &p); err != nil { c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "invalid chat_send payload")) return @@ -48,102 +50,36 @@ func (h *Hub) handleChatSend(ctx context.Context, c *Client, reqID string, paylo return } - // Check channel exists. ch, err := h.db.GetChannel(channelID) if err != nil || ch == nil { c.sendMsg(buildErrorMsg(ErrCodeNotFound, "channel not found")) return } - // DM channels use participant-based auth instead of role permissions. isDM := ch.Type == "dm" - if isDM { - ok, dmErr := h.db.IsDMParticipant(c.userID, channelID) - if dmErr != nil { - slog.Error("ws handleChatSend IsDMParticipant", "err", dmErr) - c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to check DM participation")) - return - } - if !ok { - c.sendMsg(buildErrorMsg(ErrCodeForbidden, "you are not a participant in this DM")) - return - } - } else { - // Permission check for non-DM channels. - if !h.requireChannelPerm(c, channelID, permissions.ReadMessages|permissions.SendMessages, "SEND_MESSAGES") { - return - } - } - - // Slow mode enforcement: moderators with MANAGE_MESSAGES bypass it. - // DM channels do not have slow mode. - if !isDM && ch.SlowMode > 0 && !h.hasChannelPerm(c, channelID, permissions.ManageMessages) { - slowKey := fmt.Sprintf("slow:%d:%d", c.userID, channelID) - if !h.limiter.Allow(slowKey, 1, time.Duration(ch.SlowMode)*time.Second) { - c.sendMsg(buildErrorMsg(ErrCodeSlowMode, fmt.Sprintf("channel has %ds slow mode", ch.SlowMode))) - return - } - } - - // Sanitize and validate content length. - content := sanitizer.Sanitize(p.Content) - if content == "" && len(p.Attachments) == 0 { - c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "message content cannot be empty")) + if !h.checkChatSendPermission(c, channelID, isDM) { return } - if len([]rune(content)) > maxMessageLen { - c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "message content exceeds maximum length of 4000 characters")) + if !h.checkSlowMode(c, ch, channelID, isDM) { + return + } + + content, ok := h.validateChatContent(c, p.Content, p.Attachments) + if !ok { return } - // Check attachment permission before persisting anything. - // DM channels use participant-based auth (already checked above), not role permissions. if !isDM && len(p.Attachments) > 0 { if !h.requireChannelPerm(c, channelID, permissions.AttachFiles, "ATTACH_FILES") { return } } - // Persist message. - msgID, err := h.db.CreateMessage(channelID, c.userID, content, p.ReplyTo) - if err != nil { - slog.Error("ws handleChatSend CreateMessage", "err", err) - c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to save message")) + msgID, attachments, ok := h.persistChatMessage(c, channelID, content, p.ReplyTo, p.Attachments) + if !ok { return } - // Link attachments if provided. - var attachments []map[string]any - if len(p.Attachments) > 0 { - linked, linkErr := h.db.LinkAttachmentsToMessage(msgID, p.Attachments) - if linkErr != nil { - slog.Error("ws handleChatSend LinkAttachments", "err", linkErr, "msg_id", msgID) - // Delete the orphaned message so it doesn't persist without its attachments. - if delErr := h.db.DeleteMessage(msgID, c.userID, true); delErr != nil { - slog.Error("ws handleChatSend DeleteMessage (cleanup)", "err", delErr, "msg_id", msgID) - } - c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to send message with attachments")) - return - } - if linked > 0 { - attMap, attErr := h.db.GetAttachmentsByMessageIDs([]int64{msgID}) - if attErr != nil { - slog.Error("ws handleChatSend GetAttachments", "err", attErr) - } else { - for _, ai := range attMap[msgID] { - attachments = append(attachments, map[string]any{ - "id": ai.ID, - "filename": ai.Filename, - "size": ai.Size, - "mime": ai.Mime, - "url": ai.URL, - }) - } - } - } - } - - // Retrieve to get timestamp. msg, err := h.db.GetMessage(msgID) if err != nil || msg == nil { slog.Error("ws handleChatSend GetMessage after create", "err", err) @@ -159,47 +95,136 @@ func (h *Hub) handleChatSend(ctx context.Context, c *Client, reqID string, paylo } slog.Debug("message sent", "user", username, "channel_id", channelID, "msg_id", msgID) - - // Ack sender. c.sendMsg(buildChatSendOK(reqID, msgID, msg.Timestamp)) - // Broadcast message. broadcast := buildChatMessage(msgID, channelID, c.userID, username, avatar, c.roleName, content, msg.Timestamp, p.ReplyTo, attachments) + h.broadcastChatMessage(c, channelID, isDM, broadcast) +} +func (h *Hub) checkChatSendPermission(c *Client, channelID int64, isDM bool) bool { if isDM { - // DM: send directly to both participants instead of channel broadcast. - participantIDs, pErr := h.db.GetDMParticipantIDs(channelID) - if pErr != nil { - slog.Error("ws handleChatSend GetDMParticipantIDs", "err", pErr, "channel_id", channelID) - // Message is already persisted but we cannot deliver it. Inform the - // sender so the failure is not silent. - c.sendMsg(buildErrorMsg(ErrCodeInternal, "message saved but delivery failed — please retry")) - return + ok, dmErr := h.db.IsDMParticipant(c.userID, channelID) + if dmErr != nil { + slog.Error("ws handleChatSend IsDMParticipant", "err", dmErr) + c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to check DM participation")) + return false } + if !ok { + c.sendMsg(buildErrorMsg(ErrCodeForbidden, "you are not a participant in this DM")) + return false + } + return true + } + return h.requireChannelPerm(c, channelID, permissions.ReadMessages|permissions.SendMessages, "SEND_MESSAGES") +} - // Deliver chat_message to all DM participants. - for _, pid := range participantIDs { - h.SendToUser(pid, broadcast) - } +func (h *Hub) checkSlowMode(c *Client, ch *db.Channel, channelID int64, isDM bool) bool { + if isDM || ch.SlowMode <= 0 || h.hasChannelPerm(c, channelID, permissions.ManageMessages) { + return true + } + slowKey := fmt.Sprintf("slow:%d:%d", c.userID, channelID) + if !h.limiter.Allow(slowKey, 1, time.Duration(ch.SlowMode)*time.Second) { + c.sendMsg(buildErrorMsg(ErrCodeSlowMode, fmt.Sprintf("channel has %ds slow mode", ch.SlowMode))) + return false + } + return true +} - // Auto-reopen the DM for the recipient if it was closed. - for _, pid := range participantIDs { - if pid == c.userID { - continue - } - if openErr := h.db.OpenDM(pid, channelID); openErr != nil { - slog.Error("ws handleChatSend OpenDM", "err", openErr, - "recipient_id", pid, "channel_id", channelID) - continue - } - // Notify the recipient that the DM was (re)opened. - // Build the event with the sender as the recipient's "other user". - if c.user != nil { - h.SendToUser(pid, buildDMChannelOpen(channelID, c.user)) - } +func (h *Hub) validateChatContent(c *Client, raw string, attachments []string) (string, bool) { + content := sanitizer.Sanitize(raw) + if content == "" && len(attachments) == 0 { + c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "message content cannot be empty")) + return "", false + } + if len([]rune(content)) > maxMessageLen { + c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "message content exceeds maximum length of 4000 characters")) + return "", false + } + return content, true +} + +func (h *Hub) persistChatMessage(c *Client, channelID int64, content string, replyTo *int64, attIDs []string) (int64, []map[string]any, bool) { + msgID, err := h.db.CreateMessage(channelID, c.userID, content, replyTo) + if err != nil { + slog.Error("ws handleChatSend CreateMessage", "err", err) + c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to save message")) + return 0, nil, false + } + + attachments, ok := h.linkAttachments(c, msgID, attIDs) + if !ok { + return 0, nil, false + } + return msgID, attachments, true +} + +func (h *Hub) linkAttachments(c *Client, msgID int64, attIDs []string) ([]map[string]any, bool) { + if len(attIDs) == 0 { + return nil, true + } + + linked, linkErr := h.db.LinkAttachmentsToMessage(msgID, attIDs) + if linkErr != nil { + slog.Error("ws handleChatSend LinkAttachments", "err", linkErr, "msg_id", msgID) + if delErr := h.db.DeleteMessage(msgID, c.userID, true); delErr != nil { + slog.Error("ws handleChatSend DeleteMessage (cleanup)", "err", delErr, "msg_id", msgID) } - } else { + c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to send message with attachments")) + return nil, false + } + + if linked == 0 { + return nil, true + } + + attMap, attErr := h.db.GetAttachmentsByMessageIDs([]int64{msgID}) + if attErr != nil { + slog.Error("ws handleChatSend GetAttachments", "err", attErr) + return nil, true + } + + var attachments []map[string]any + for _, ai := range attMap[msgID] { + attachments = append(attachments, map[string]any{ + "id": ai.ID, + "filename": ai.Filename, + "size": ai.Size, + "mime": ai.Mime, + "url": ai.URL, + }) + } + return attachments, true +} + +func (h *Hub) broadcastChatMessage(c *Client, channelID int64, isDM bool, broadcast []byte) { + if !isDM { h.BroadcastToChannel(channelID, broadcast) + return + } + + participantIDs, pErr := h.db.GetDMParticipantIDs(channelID) + if pErr != nil { + slog.Error("ws handleChatSend GetDMParticipantIDs", "err", pErr, "channel_id", channelID) + c.sendMsg(buildErrorMsg(ErrCodeInternal, "message saved but delivery failed — please retry")) + return + } + + for _, pid := range participantIDs { + h.SendToUser(pid, broadcast) + } + + for _, pid := range participantIDs { + if pid == c.userID { + continue + } + if openErr := h.db.OpenDM(pid, channelID); openErr != nil { + slog.Error("ws handleChatSend OpenDM", "err", openErr, + "recipient_id", pid, "channel_id", channelID) + continue + } + if c.user != nil { + h.SendToUser(pid, buildDMChannelOpen(channelID, c.user)) + } } } diff --git a/Server/ws/hub_test.go b/Server/ws/hub_test.go index 4d629023..2f852c85 100644 --- a/Server/ws/hub_test.go +++ b/Server/ws/hub_test.go @@ -518,6 +518,129 @@ func TestHub_CleanupVoiceForChannel_NoVoiceState_NoPanic(t *testing.T) { // the client map entry; voice cleanup for disconnects is handled by // handleVoiceLeave called from readPump/ICE monitor. +// ─── sweepStaleClients ────────────────────────────────────────────────────── + +func TestHub_SweepStaleClients_RemovesInactiveClients(t *testing.T) { + hub, database := newTestHub(t) + go hub.Run() + defer hub.Stop() + + u1 := seedTestUser(t, database, "stale-alice") + u2 := seedTestUser(t, database, "fresh-bob") + + s1 := make(chan []byte, 4) + s2 := make(chan []byte, 4) + c1 := ws.NewTestClient(hub, u1, s1) + c2 := ws.NewTestClient(hub, u2, s2) + + hub.Register(c1) + hub.Register(c2) + time.Sleep(20 * time.Millisecond) + + ws.SetClientLastActivityForTest(c1, time.Now().Add(-2*time.Minute)) + ws.SetClientLastActivityForTest(c2, time.Now()) + + hub.SweepStaleClientsForTest() + time.Sleep(20 * time.Millisecond) + + if hub.ClientCount() != 1 { + t.Errorf("ClientCount = %d after sweep, want 1", hub.ClientCount()) + } + if hub.GetClient(u1) != nil { + t.Error("stale client should have been removed") + } + if hub.GetClient(u2) == nil { + t.Error("fresh client should still be present") + } +} + +func TestHub_SweepStaleClients_NoClientsNoPanic(t *testing.T) { + hub, _ := newTestHub(t) + hub.SweepStaleClientsForTest() +} + +func TestHub_SweepStaleClients_AllFresh(t *testing.T) { + hub, database := newTestHub(t) + go hub.Run() + defer hub.Stop() + + u1 := seedTestUser(t, database, "fresh-carol") + s1 := make(chan []byte, 4) + c1 := ws.NewTestClient(hub, u1, s1) + hub.Register(c1) + time.Sleep(20 * time.Millisecond) + + ws.SetClientLastActivityForTest(c1, time.Now()) + hub.SweepStaleClientsForTest() + time.Sleep(20 * time.Millisecond) + + if hub.ClientCount() != 1 { + t.Errorf("ClientCount = %d after sweep of fresh clients, want 1", hub.ClientCount()) + } +} + +// ─── LiveKitHealthCheck ───────────────────────────────────────────────────── + +func TestHub_LiveKitHealthCheck_NilReturnsError(t *testing.T) { + hub, _ := newTestHub(t) + ok, err := hub.LiveKitHealthCheck() + if ok { + t.Error("expected ok=false when LiveKit is nil") + } + if err == nil { + t.Error("expected non-nil error when LiveKit is nil") + } +} + +// ─── SetLiveKitProcess ────────────────────────────────────────────────────── + +func TestHub_SetLiveKitProcess(t *testing.T) { + hub, _ := newTestHub(t) + hub.SetLiveKitProcess(nil) + go hub.Run() + hub.GracefulStop() +} + +// ─── VoiceSessionCount ───────────────────────────────────────────────────── + +func TestHub_VoiceSessionCount(t *testing.T) { + tests := []struct { + name string + voiceChs []int64 + wantCount int + }{ + {"no clients", nil, 0}, + {"all in voice", []int64{100, 200, 300}, 3}, + {"none in voice", []int64{0, 0}, 0}, + {"mixed", []int64{100, 0, 200, 0}, 2}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + hub, database := newTestHub(t) + go hub.Run() + defer hub.Stop() + + for i, vch := range tc.voiceChs { + username := fmt.Sprintf("voice-%s-%d", tc.name, i) + uid := seedTestUser(t, database, username) + send := make(chan []byte, 4) + c := ws.NewTestClient(hub, uid, send) + if vch != 0 { + ws.SetClientVoiceChID(c, vch) + } + hub.Register(c) + } + time.Sleep(30 * time.Millisecond) + + got := hub.VoiceSessionCount() + if got != tc.wantCount { + t.Errorf("VoiceSessionCount() = %d, want %d", got, tc.wantCount) + } + }) + } +} + // hubTestSchema is the minimal schema needed for hub tests. var hubTestSchema = []byte(` CREATE TABLE IF NOT EXISTS roles ( diff --git a/Server/ws/livekit_test.go b/Server/ws/livekit_test.go index ed1940e8..cb95cd89 100644 --- a/Server/ws/livekit_test.go +++ b/Server/ws/livekit_test.go @@ -2,13 +2,19 @@ package ws_test import ( "encoding/json" + "net/http" + "net/http/httptest" + "os" + "strings" "testing" "time" + "github.com/go-chi/chi/v5" "github.com/owncord/server/config" "github.com/owncord/server/ws" ) + // --------------------------------------------------------------------------- // livekit.go tests // --------------------------------------------------------------------------- @@ -508,3 +514,351 @@ func TestWebhook_ParticipantLeft_OldToken_DoesNotTeardownReplacement(t *testing. t.Fatalf("replacement join token = %q, want %q", vs.JoinedAt, newState.JoinedAt) } } + +// --------------------------------------------------------------------------- +// livekit_process.go – generateConfig tests +// --------------------------------------------------------------------------- + +func TestGenerateConfig_WritesYAML(t *testing.T) { + t.Parallel() + + dataDir := t.TempDir() + cfg := &config.VoiceConfig{ + LiveKitAPIKey: "testkey", + LiveKitAPISecret: "testsecret", + LiveKitURL: "ws://localhost:7880", + } + tlsCfg := &config.TLSConfig{} + + proc := ws.NewLiveKitProcess(cfg, tlsCfg, dataDir) + + cfgPath, err := proc.GenerateConfigForTest() + if err != nil { + t.Fatalf("generateConfig: %v", err) + } + + content, err := os.ReadFile(cfgPath) + if err != nil { + t.Fatalf("reading config file: %v", err) + } + + got := string(content) + + for _, want := range []string{ + "port: 7880", + `"testkey": "testsecret"`, + "port_range_start: 50000", + "port_range_end: 60000", + "use_external_ip: true", + "level: info", + } { + if !strings.Contains(got, want) { + t.Errorf("config missing %q.\nGot:\n%s", want, got) + } + } + + if strings.Contains(got, "node_ip") { + t.Error("config should not contain node_ip when NodeIP is empty") + } +} + +func TestGenerateConfig_WithNodeIP(t *testing.T) { + t.Parallel() + + dataDir := t.TempDir() + cfg := &config.VoiceConfig{ + LiveKitAPIKey: "key1", + LiveKitAPISecret: "secret1", + LiveKitURL: "ws://localhost:7880", + NodeIP: "203.0.113.10", + } + tlsCfg := &config.TLSConfig{} + + proc := ws.NewLiveKitProcess(cfg, tlsCfg, dataDir) + + cfgPath, err := proc.GenerateConfigForTest() + if err != nil { + t.Fatalf("generateConfig: %v", err) + } + + content, err := os.ReadFile(cfgPath) + if err != nil { + t.Fatalf("reading config file: %v", err) + } + + got := string(content) + if !strings.Contains(got, `node_ip: "203.0.113.10"`) { + t.Errorf("expected node_ip in config.\nGot:\n%s", got) + } +} + +func TestGenerateConfig_UnsafeCredentialChars(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + key string + secret string + }{ + {"colon in key", "bad:key", "secret"}, + {"newline in secret", "key", "bad\nsecret"}, + {"hash in key", "bad#key", "secret"}, + {"brace in secret", "key", "bad{secret"}, + {"backslash in key", `bad\key`, "secret"}, + {"quote in secret", "key", `bad"secret`}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + cfg := &config.VoiceConfig{ + LiveKitAPIKey: tt.key, + LiveKitAPISecret: tt.secret, + LiveKitURL: "ws://localhost:7880", + } + tlsCfg := &config.TLSConfig{} + proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir()) + + _, err := proc.GenerateConfigForTest() + if err == nil { + t.Error("expected error for unsafe YAML character, got nil") + } + }) + } +} + +func TestGenerateConfig_UnsafeNodeIPChars(t *testing.T) { + t.Parallel() + + cfg := &config.VoiceConfig{ + LiveKitAPIKey: "safekey", + LiveKitAPISecret: "safesecret", + LiveKitURL: "ws://localhost:7880", + NodeIP: "192.168.1.1\n evil: true", + } + tlsCfg := &config.TLSConfig{} + proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir()) + + _, err := proc.GenerateConfigForTest() + if err == nil { + t.Error("expected error for unsafe node_ip character, got nil") + } +} + +// --------------------------------------------------------------------------- +// livekit_process.go – Start guard tests +// --------------------------------------------------------------------------- + +func TestStart_AlreadyRunningGuard(t *testing.T) { + t.Parallel() + + cfg := &config.VoiceConfig{ + LiveKitAPIKey: "key", + LiveKitAPISecret: "secret", + LiveKitURL: "ws://localhost:7880", + LiveKitBinaryPath: "/nonexistent/livekit-server", + } + tlsCfg := &config.TLSConfig{} + proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir()) + + proc.SetProcessCmdForTest() + + err := proc.Start() + if err == nil { + t.Fatal("expected error when process already running, got nil") + } + if !strings.Contains(err.Error(), "already running") { + t.Errorf("expected 'already running' error, got: %v", err) + } +} + +func TestStart_StoppedGuard(t *testing.T) { + t.Parallel() + + cfg := &config.VoiceConfig{ + LiveKitAPIKey: "key", + LiveKitAPISecret: "secret", + LiveKitURL: "ws://localhost:7880", + LiveKitBinaryPath: "/nonexistent/livekit-server", + } + tlsCfg := &config.TLSConfig{} + proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir()) + + proc.SetProcessStoppedForTest() + + err := proc.Start() + if err != nil { + t.Fatalf("Start() on stopped process returned error: %v", err) + } + + proc.Stop() + if proc.IsRunning() { + t.Error("expected IsRunning() = false after Stop()") + } +} + +// --------------------------------------------------------------------------- +// livekit_process.go – HealthCheck tests +// --------------------------------------------------------------------------- + +func TestHealthCheck_Success(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + wsURL := "ws://" + srv.Listener.Addr().String() + + cfg := &config.VoiceConfig{ + LiveKitAPIKey: "key", + LiveKitAPISecret: "secret", + LiveKitURL: wsURL, + } + tlsCfg := &config.TLSConfig{} + + proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir()) + + ok, err := proc.HealthCheck() + if err != nil { + t.Fatalf("HealthCheck: %v", err) + } + if !ok { + t.Error("expected HealthCheck to return true") + } +} + +func TestHealthCheck_ServerDown(t *testing.T) { + t.Parallel() + + cfg := &config.VoiceConfig{ + LiveKitAPIKey: "key", + LiveKitAPISecret: "secret", + LiveKitURL: "ws://127.0.0.1:1", + } + tlsCfg := &config.TLSConfig{} + + proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir()) + + ok, err := proc.HealthCheck() + if err == nil { + t.Fatal("expected error for unreachable server, got nil") + } + if ok { + t.Error("expected HealthCheck to return false on error") + } +} + +func TestHealthCheck_NonOKStatus(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + wsURL := "ws://" + srv.Listener.Addr().String() + cfg := &config.VoiceConfig{ + LiveKitAPIKey: "key", + LiveKitAPISecret: "secret", + LiveKitURL: wsURL, + } + tlsCfg := &config.TLSConfig{} + proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir()) + + ok, err := proc.HealthCheck() + if err != nil { + t.Fatalf("HealthCheck: %v", err) + } + if !ok { + t.Error("expected HealthCheck to return true even for 500 status") + } +} + +// --------------------------------------------------------------------------- +// livekit_webhook.go – NewLiveKitWebhookHandler tests +// --------------------------------------------------------------------------- + +func TestWebhookHandler_MissingAuthHeader(t *testing.T) { + t.Parallel() + + hub := ws.NewHubForTest() + handler := hub.NewLiveKitWebhookHandler("api-key", "api-secret") + + req := httptest.NewRequest(http.MethodPost, "/livekit/webhook", strings.NewReader(`{}`)) + rec := httptest.NewRecorder() + + handler(rec, req) + + if rec.Code != http.StatusUnauthorized { + t.Errorf("expected 401, got %d", rec.Code) + } +} + +func TestWebhookHandler_InvalidToken(t *testing.T) { + t.Parallel() + + hub := ws.NewHubForTest() + handler := hub.NewLiveKitWebhookHandler("api-key", "api-secret") + + req := httptest.NewRequest(http.MethodPost, "/livekit/webhook", + strings.NewReader(`{}`)) + req.Header.Set("Authorization", "Bearer not-a-valid-jwt-token") + rec := httptest.NewRecorder() + + handler(rec, req) + + if rec.Code != http.StatusUnauthorized { + t.Errorf("expected 401 for invalid token, got %d", rec.Code) + } +} + +func TestWebhookHandler_EmptyBody(t *testing.T) { + t.Parallel() + + hub := ws.NewHubForTest() + handler := hub.NewLiveKitWebhookHandler("api-key", "api-secret") + + req := httptest.NewRequest(http.MethodPost, "/livekit/webhook", + strings.NewReader("")) + rec := httptest.NewRecorder() + + handler(rec, req) + + if rec.Code != http.StatusUnauthorized { + t.Errorf("expected 401 for missing auth, got %d", rec.Code) + } +} + +// --------------------------------------------------------------------------- +// livekit_webhook.go – MountWebhookRoute tests +// --------------------------------------------------------------------------- + +func TestMountWebhookRoute_RegistersRoute(t *testing.T) { + t.Parallel() + + hub := ws.NewHubForTest() + handler := ws.MountWebhookRoute(hub, "key", "secret") + + if handler == nil { + t.Fatal("MountWebhookRoute returned nil handler") + } + + r := chi.NewRouter() + r.Post("/livekit/webhook", handler) + + req := httptest.NewRequest(http.MethodPost, "/livekit/webhook", + strings.NewReader(`{}`)) + rec := httptest.NewRecorder() + + r.ServeHTTP(rec, req) + + if rec.Code == http.StatusNotFound { + t.Error("expected route to be registered, got 404") + } + if rec.Code != http.StatusUnauthorized { + t.Errorf("expected 401 from mounted webhook handler, got %d", rec.Code) + } +} diff --git a/Server/ws/serve.go b/Server/ws/serve.go index 88073e79..c0744a47 100644 --- a/Server/ws/serve.go +++ b/Server/ws/serve.go @@ -36,39 +36,12 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun } conn.SetReadLimit(1 << 20) // 1 MB — match client-side limit - user, tokenHash, lastSeq, err := authenticateConn(conn, database) + c, lastSeq, err := hub.upgradeAndAuth(conn, database, r) if err != nil { - slog.Warn("ws auth failed", "err", err, "remote", r.RemoteAddr) - _ = conn.Close(websocket.StatusPolicyViolation, "authentication failed") return } - c := newClient(hub, conn, user, tokenHash, lastSeq, r.Context()) - c.remoteAddr = r.RemoteAddr - - // Look up role name for protocol-compliant payloads and cache on client. - roleName := "member" - if role, roleErr := database.GetRoleByID(user.RoleID); roleErr == nil && role != nil { - roleName = strings.ToLower(role.Name) - } - c.roleName = roleName - - slog.Info("websocket connected", "username", user.Username, "user_id", user.ID, "remote", r.RemoteAddr) - _ = database.LogAudit(user.ID, "ws_connect", "user", user.ID, - "WebSocket connected from "+r.RemoteAddr) - ctx := r.Context() - hydrateVoiceJoinToken := func() { - voiceChID := c.getVoiceChID() - if voiceChID == 0 || c.getVoiceJoinToken() != "" { - return - } - vs, vsErr := database.GetVoiceState(user.ID) - if vsErr != nil || vs == nil || vs.ChannelID != voiceChID || vs.JoinedAt == "" { - return - } - c.setVoiceState(voiceChID, vs.JoinedAt) - } startPumps := func() { writeCtx, writeCancel := context.WithCancel(ctx) go writePump(writeCtx, conn, c) @@ -81,130 +54,17 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun // try to replay missed events from the ring buffer instead of // sending a full ready payload. if lastSeq > 0 { - events := hub.ReplayBuffer().EventsSince(lastSeq) - if events != nil { - // Replay succeeded — send auth_ok then missed events. - slog.Info("ws sending auth_ok (reconnect)", "user_id", user.ID, "username", user.Username, "role", roleName) - if err := conn.Write(ctx, websocket.MessageText, hub.buildAuthOK(user, roleName)); err != nil { - slog.Warn("ws: failed to send auth_ok (reconnect)", "user_id", user.ID, "err", err) - _ = conn.Close(websocket.StatusInternalError, "handshake failed") - return - } - for _, evt := range events { - if err := conn.Write(ctx, websocket.MessageText, evt); err != nil { - slog.Warn("ws: failed to send replay event", "user_id", user.ID, "err", err) - _ = conn.Close(websocket.StatusInternalError, "handshake failed") - return - } - } - slog.Info("ws replay completed", "user_id", user.ID, "events_replayed", len(events), "from_seq", lastSeq) - hub.registerNow(c) - hydrateVoiceJoinToken() - - // Update presence but skip member_join — user was already known. - if updateErr := database.UpdateUserStatus(user.ID, "online"); updateErr != nil { - slog.Warn("ws UpdateUserStatus", "err", updateErr) - } - hub.BroadcastToAll(buildPresenceMsg(user.ID, "online")) - - // Start pumps. + if hub.handleReconnect(ctx, conn, c, database, lastSeq) { startPumps() return } // Replay failed (seq too old) — fall through to full ready payload. - slog.Info("ws replay failed (seq too old), sending full ready", "user_id", user.ID, "last_seq", lastSeq) + slog.Info("ws replay failed (seq too old), sending full ready", "user_id", c.userID, "last_seq", lastSeq) } - // Fresh connection or replay fallback: full auth_ok + ready flow. - - // Clean any stale voice state BEFORE building the ready payload so - // the user doesn't appear as a ghost in a voice channel they left - // abruptly (e.g. F5 reload). Only for truly fresh connections - // (lastSeq == 0); when lastSeq > 0 the client still has its - // JS context with a LiveKit room — it just needs a new ready payload - // because the replay buffer was too old. - // - // This is the SINGLE authoritative cleanup path for fresh connections. - // registerNow does NOT duplicate this — it only handles in-memory - // client replacement and voice state transfer for lastSeq > 0. - if lastSeq == 0 { - vs, vsErr := database.GetVoiceState(user.ID) - if vsErr != nil { - // DB read failure — fail closed. A transient read failure - // could leak stale voice state into the ready payload. - slog.Error("ws: GetVoiceState failed — aborting connection", - "user_id", user.ID, "err", vsErr) - _ = conn.Write(ctx, websocket.MessageText, - buildErrorMsg(ErrCodeInternal, "voice state check failed")) - _ = conn.Close(websocket.StatusInternalError, "voice state check failed") - return - } - if vs != nil { - staleChID := vs.ChannelID - slog.Info("ws cleaning stale voice state before ready", - "user_id", user.ID, "stale_channel_id", staleChID) - // Channel-conditional delete: only removes the row if it still - // points at staleChID. If the old connection moved the user to - // a different channel between GetVoiceState and now, the delete - // is a safe no-op and we skip the broadcast. - deleted, dbErr := database.LeaveVoiceChannelIfMatch(user.ID, staleChID, vs.JoinedAt) - if dbErr != nil { - slog.Error("ws: stale voice cleanup failed — aborting connection", - "user_id", user.ID, "channel_id", staleChID, "err", dbErr) - _ = conn.Write(ctx, websocket.MessageText, - buildErrorMsg(ErrCodeInternal, "voice state cleanup failed")) - _ = conn.Close(websocket.StatusInternalError, "voice cleanup failed") - return - } - if deleted { - // Clear the old client's in-memory voice state BEFORE - // calling RemoveParticipant. RemoveParticipant triggers a - // LiveKit participant_left webhook; if the old client - // still carries the matching join token, the webhook - // handler's token-match branch would broadcast a second - // voice_leave. Clearing first makes that branch a no-op. - hub.mu.RLock() - if oldClient, ok := hub.clients[user.ID]; ok { - oldClient.clearVoiceState() - } - hub.mu.RUnlock() - - hub.BroadcastToAll(buildVoiceLeave(staleChID, user.ID)) - if hub.livekit != nil { - _ = hub.livekit.RemoveParticipant(staleChID, user.ID, vs.JoinedAt) - } - } - } - } - - slog.Info("ws sending auth_ok", "user_id", user.ID, "username", user.Username, "role", roleName) - if err := conn.Write(ctx, websocket.MessageText, hub.buildAuthOK(user, roleName)); err != nil { - slog.Warn("ws: failed to send auth_ok", "user_id", user.ID, "err", err) - _ = conn.Close(websocket.StatusInternalError, "handshake failed") + if err := hub.handleFreshConnect(ctx, conn, c, database); err != nil { return } - if ready, readyErr := hub.buildReady(database, user.ID); readyErr == nil { - slog.Info("ws sending ready payload", "user_id", user.ID, "payload_bytes", len(ready)) - if err := conn.Write(ctx, websocket.MessageText, ready); err != nil { - slog.Warn("ws: failed to send ready payload", "user_id", user.ID, "err", err) - _ = conn.Close(websocket.StatusInternalError, "handshake failed") - return - } - } else { - slog.Error("buildReady failed", "user_id", user.ID, "err", readyErr) - _ = conn.Write(ctx, websocket.MessageText, - buildErrorMsg(ErrCodeInternal, "failed to build ready payload")) - } - hub.registerNow(c) - hydrateVoiceJoinToken() - - if updateErr := database.UpdateUserStatus(user.ID, "online"); updateErr != nil { - slog.Warn("ws UpdateUserStatus", "err", updateErr) - } - - slog.Info("ws broadcasting member_join and presence", "user_id", user.ID, "username", user.Username) - hub.BroadcastToAll(buildMemberJoin(user, roleName)) - hub.BroadcastToAll(buildPresenceMsg(user.ID, "online")) // writePump runs in background; readPump blocks. // When readPump returns (disconnect), close the send channel first @@ -213,6 +73,102 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun } } +func (h *Hub) upgradeAndAuth( + conn *websocket.Conn, database *db.DB, r *http.Request, +) (*Client, uint64, error) { + user, tokenHash, lastSeq, err := authenticateConn(conn, database) + if err != nil { + slog.Warn("ws auth failed", "err", err, "remote", r.RemoteAddr) + _ = conn.Close(websocket.StatusPolicyViolation, "authentication failed") + return nil, 0, err + } + + c := newClient(h, conn, user, tokenHash, lastSeq, r.Context()) + c.remoteAddr = r.RemoteAddr + + // Look up role name for protocol-compliant payloads and cache on client. + roleName := "member" + if role, roleErr := database.GetRoleByID(user.RoleID); roleErr == nil && role != nil { + roleName = strings.ToLower(role.Name) + } + c.roleName = roleName + + slog.Info("websocket connected", "username", user.Username, "user_id", user.ID, "remote", r.RemoteAddr) + _ = database.LogAudit(user.ID, "ws_connect", "user", user.ID, + "WebSocket connected from "+r.RemoteAddr) + + return c, lastSeq, nil +} + +func (h *Hub) handleReconnect( + ctx context.Context, conn *websocket.Conn, c *Client, database *db.DB, lastSeq uint64, +) bool { + events := h.ReplayBuffer().EventsSince(lastSeq) + if events == nil { + return false + } + + // Replay succeeded — send auth_ok then missed events. + slog.Info("ws sending auth_ok (reconnect)", "user_id", c.userID, "username", c.user.Username, "role", c.roleName) + if err := conn.Write(ctx, websocket.MessageText, h.buildAuthOK(c.user, c.roleName)); err != nil { + slog.Warn("ws: failed to send auth_ok (reconnect)", "user_id", c.userID, "err", err) + _ = conn.Close(websocket.StatusInternalError, "handshake failed") + return true + } + for _, evt := range events { + if err := conn.Write(ctx, websocket.MessageText, evt); err != nil { + slog.Warn("ws: failed to send replay event", "user_id", c.userID, "err", err) + _ = conn.Close(websocket.StatusInternalError, "handshake failed") + return true + } + } + slog.Info("ws replay completed", "user_id", c.userID, "events_replayed", len(events), "from_seq", lastSeq) + h.registerNow(c) + + // Update presence but skip member_join — user was already known. + if updateErr := database.UpdateUserStatus(c.userID, "online"); updateErr != nil { + slog.Warn("ws UpdateUserStatus", "err", updateErr) + } + h.BroadcastToAll(buildPresenceMsg(c.userID, "online")) + + return true +} + +func (h *Hub) handleFreshConnect( + ctx context.Context, conn *websocket.Conn, c *Client, database *db.DB, +) error { + // Fresh connection or replay fallback: full auth_ok + ready flow. + slog.Info("ws sending auth_ok", "user_id", c.userID, "username", c.user.Username, "role", c.roleName) + if err := conn.Write(ctx, websocket.MessageText, h.buildAuthOK(c.user, c.roleName)); err != nil { + slog.Warn("ws: failed to send auth_ok", "user_id", c.userID, "err", err) + _ = conn.Close(websocket.StatusInternalError, "handshake failed") + return err + } + if ready, readyErr := h.buildReady(database, c.userID); readyErr == nil { + slog.Info("ws sending ready payload", "user_id", c.userID, "payload_bytes", len(ready)) + if err := conn.Write(ctx, websocket.MessageText, ready); err != nil { + slog.Warn("ws: failed to send ready payload", "user_id", c.userID, "err", err) + _ = conn.Close(websocket.StatusInternalError, "handshake failed") + return err + } + } else { + slog.Error("buildReady failed", "user_id", c.userID, "err", readyErr) + _ = conn.Write(ctx, websocket.MessageText, + buildErrorMsg(ErrCodeInternal, "failed to build ready payload")) + } + h.registerNow(c) + + if updateErr := database.UpdateUserStatus(c.userID, "online"); updateErr != nil { + slog.Warn("ws UpdateUserStatus", "err", updateErr) + } + + slog.Info("ws broadcasting member_join and presence", "user_id", c.userID, "username", c.user.Username) + h.BroadcastToAll(buildMemberJoin(c.user, c.roleName)) + h.BroadcastToAll(buildPresenceMsg(c.userID, "online")) + + return nil +} + // writePump drains the client's send channel and writes to the WebSocket. func writePump(ctx context.Context, conn *websocket.Conn, c *Client) { for { diff --git a/docs/api.md b/docs/api.md index 7ca9bd98..eb420c84 100644 --- a/docs/api.md +++ b/docs/api.md @@ -248,6 +248,36 @@ Invalidate the current session token. --- +### DELETE /api/v1/auth/account + +Permanently delete the authenticated user's account. Requires password confirmation. + +**Auth:** Required (Bearer token) +**Rate limit:** 5 requests/minute per IP. After 3 failed password attempts, the endpoint locks out for 15 minutes per user. + +#### Request + +```json +{ + "password": "MyStr0ng!Pass" +} +``` + +#### Response 204 No Content + +Account deleted successfully. All sessions, messages (soft-deleted), and associated data are cleaned up. + +#### Errors + +| Status | Code | Cause | +| ------ | ---- | ----- | +| 400 | `INVALID_INPUT` | Missing or incorrect password | +| 403 | `FORBIDDEN` | Cannot delete the last admin account | +| 429 | `RATE_LIMITED` | Locked out after 3 failed password attempts (15 min cooldown) | +| 500 | `SERVER_ERROR` | Database error during deletion | + +--- + ### POST /api/v1/users/me/totp/enable Start TOTP enrollment for the authenticated user. The secret is not persisted until `/api/v1/users/me/totp/confirm` succeeds. diff --git a/docs/contributing.md b/docs/contributing.md index bdf1af78..ea599e49 100644 --- a/docs/contributing.md +++ b/docs/contributing.md @@ -7,7 +7,7 @@ How to set up the development environment and contribute to OwnCord. ### Prerequisites - **Windows 10+** (x64) -- **Go 1.22+** (server) +- **Go 1.25+** (server) - **Node.js 20+** (client) - **Rust / Cargo** (Tauri client) @@ -40,6 +40,7 @@ How to set up the development environment and contribute to OwnCord. | `npm run test:watch` | Vitest watch mode | | `npm run test:coverage` | Coverage report | | `npm run typecheck` | Full typecheck (all sources) | +| `npm run typecheck:build` | Typecheck build config only | | `npm run lint` | ESLint check (src/) | | `npm run lint:fix` | ESLint auto-fix | diff --git a/docs/deployment.md b/docs/deployment.md index aa807dcf..178ce2ba 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -5,7 +5,7 @@ Production deployment guide for OwnCord server on Windows. ## Prerequisites - **Windows 10+** (x64) -- **Go 1.22+** (only if building from source) +- **Go 1.25+** (only if building from source) - **LiveKit Server** binary (for voice/video) -- see [LiveKit Setup](livekit-setup.md) - Ports available: `8443` (default), `7880` (LiveKit), `80` (if using ACME/Let's Encrypt) @@ -115,10 +115,10 @@ The database uses SQLite WAL mode. Do NOT copy the `.db` file directly while the | Endpoint | Method | Description | |----------|--------|-------------| -| `/admin/api/backups` | POST | Create a new backup | +| `/admin/api/backup` | POST | Create a new backup (owner-only) | | `/admin/api/backups` | GET | List all backups (newest first) | -| `/admin/api/backups/{name}` | DELETE | Delete a backup | -| `/admin/api/backups/{name}/restore` | POST | Restore from backup (creates pre-restore safety backup first) | +| `/admin/api/backups/{name}` | DELETE | Delete a backup (owner-only) | +| `/admin/api/backups/{name}/restore` | POST | Restore from backup (owner-only; creates pre-restore safety backup first) | Backups are stored in `data/backups/` with timestamps. @@ -128,7 +128,7 @@ Use Windows Task Scheduler with PowerShell: ```powershell $headers = @{ "Cookie" = "session=" } -Invoke-RestMethod -Uri "https://localhost:8443/admin/api/backups" -Method POST -Headers $headers -SkipCertificateCheck +Invoke-RestMethod -Uri "https://localhost:8443/admin/api/backup" -Method POST -Headers $headers -SkipCertificateCheck ``` ### Restore diff --git a/docs/security.md b/docs/security.md index 46cf3f2f..89414111 100644 --- a/docs/security.md +++ b/docs/security.md @@ -23,6 +23,21 @@ OwnCord supports TOTP-based 2FA: - `require_2fa` requires all users to have 2FA enabled and registration to be closed - Login flow returns `requires_2fa: true` with a `partial_token` (10-min TTL, 5-attempt limit) - Auth challenges are rate-limited to 10 req/min per IP +- TOTP code verification uses constant-time comparison (`subtle.ConstantTimeCompare`) to prevent timing side-channel attacks + +## Account Deletion + +Users can delete their own account via `DELETE /api/v1/auth/account` with password confirmation. The last admin account cannot be deleted. After 3 failed password attempts, the endpoint locks out for 15 minutes. + +## Audit Logging + +Security-relevant actions are recorded in the `audit_log` table with actor, action, target, and detail: + +- **Auth:** `user_register`, `user_login`, `user_logout`, `login_blocked_banned`, `account_deleted` +- **2FA:** `totp_enabled`, `totp_verified`, `totp_disabled` +- **Admin:** `role_change`, `user_ban`, `user_unban`, `force_logout`, `setting_change`, `server_setup` +- **Content:** `channel_create`, `channel_update`, `channel_delete`, `message_delete` +- **Ops:** `backup_create`, `backup_delete`, `backup_restore`, `ws_connect` ## Known Limitations