From 41c6dbf030b788d10e3f1f0822d959f7e316173d Mon Sep 17 00:00:00 2001 From: Naitik Date: Wed, 14 Oct 2026 14:47:24 +0530 Subject: [PATCH] fix: scope job-status WebSocket events to the job owner Broadcasting every job status to all authenticated sockets caused other users to toast and refetch. Bind connections to the session UUID and only deliver events for matching ClientID. Fixes #473 --- backend/controllers/add_task.go | 3 +- backend/controllers/complete_task.go | 3 +- backend/controllers/complete_tasks.go | 3 +- backend/controllers/delete_task.go | 3 +- backend/controllers/delete_tasks.go | 3 +- backend/controllers/edit_task.go | 3 +- backend/controllers/job_queue.go | 25 ++-- backend/controllers/modify_task.go | 3 +- backend/controllers/websocket.go | 68 +++++++-- backend/controllers/websocket_test.go | 195 ++++++++++++++++++++++++++ 10 files changed, 278 insertions(+), 31 deletions(-) create mode 100644 backend/controllers/websocket_test.go diff --git a/backend/controllers/add_task.go b/backend/controllers/add_task.go index 60ff42c0..d9dd0f95 100644 --- a/backend/controllers/add_task.go +++ b/backend/controllers/add_task.go @@ -71,7 +71,8 @@ func AddTaskHandler(w http.ResponseWriter, r *http.Request) { logStore := models.GetLogStore() job := Job{ - Name: "Add Task", + Name: "Add Task", + ClientID: requestBody.UUID, Execute: func() error { logStore.AddLog("INFO", fmt.Sprintf("Adding task: %s", requestBody.Description), requestBody.UUID, "Add Task") err := tw.AddTaskToTaskwarrior(requestBody, dueDateStr) diff --git a/backend/controllers/complete_task.go b/backend/controllers/complete_task.go index 63ddffb1..70361d38 100644 --- a/backend/controllers/complete_task.go +++ b/backend/controllers/complete_task.go @@ -55,7 +55,8 @@ func CompleteTaskHandler(w http.ResponseWriter, r *http.Request) { // } logStore := models.GetLogStore() job := Job{ - Name: "Complete Task", + Name: "Complete Task", + ClientID: uuid, Execute: func() error { logStore.AddLog("INFO", fmt.Sprintf("Completing task UUID: %s", taskuuid), uuid, "Complete Task") err := tw.CompleteTaskInTaskwarrior(email, encryptionSecret, uuid, taskuuid) diff --git a/backend/controllers/complete_tasks.go b/backend/controllers/complete_tasks.go index d9dc99b9..89512281 100644 --- a/backend/controllers/complete_tasks.go +++ b/backend/controllers/complete_tasks.go @@ -53,7 +53,8 @@ func BulkCompleteTaskHandler(w http.ResponseWriter, r *http.Request) { logStore := models.GetLogStore() job := Job{ - Name: "Bulk Complete Tasks", + Name: "Bulk Complete Tasks", + ClientID: uuid, Execute: func() error { logStore.AddLog("INFO", fmt.Sprintf("[Bulk Complete] Starting %d tasks", len(taskUUIDs)), uuid, "Bulk Complete Task") diff --git a/backend/controllers/delete_task.go b/backend/controllers/delete_task.go index 34fcd628..d958019c 100644 --- a/backend/controllers/delete_task.go +++ b/backend/controllers/delete_task.go @@ -49,7 +49,8 @@ func DeleteTaskHandler(w http.ResponseWriter, r *http.Request) { logStore := models.GetLogStore() job := Job{ - Name: "Delete Task", + Name: "Delete Task", + ClientID: uuid, Execute: func() error { logStore.AddLog("INFO", fmt.Sprintf("Deleting task UUID: %s", taskuuid), uuid, "Delete Task") err := tw.DeleteTaskInTaskwarrior(email, encryptionSecret, uuid, taskuuid) diff --git a/backend/controllers/delete_tasks.go b/backend/controllers/delete_tasks.go index f1f51cea..857206ab 100644 --- a/backend/controllers/delete_tasks.go +++ b/backend/controllers/delete_tasks.go @@ -53,7 +53,8 @@ func BulkDeleteTaskHandler(w http.ResponseWriter, r *http.Request) { logStore := models.GetLogStore() job := Job{ - Name: "Bulk Delete Tasks", + Name: "Bulk Delete Tasks", + ClientID: uuid, Execute: func() error { logStore.AddLog("INFO", fmt.Sprintf("[Bulk Delete] Starting %d tasks", len(taskUUIDs)), uuid, "Bulk Delete Task") diff --git a/backend/controllers/edit_task.go b/backend/controllers/edit_task.go index f76e8467..a4156ea2 100644 --- a/backend/controllers/edit_task.go +++ b/backend/controllers/edit_task.go @@ -118,7 +118,8 @@ func EditTaskHandler(w http.ResponseWriter, r *http.Request) { logStore := models.GetLogStore() job := Job{ - Name: "Edit Task", + Name: "Edit Task", + ClientID: uuid, Execute: func() error { logStore.AddLog("INFO", fmt.Sprintf("Editing task UUID: %s", taskUUID), uuid, "Edit Task") diff --git a/backend/controllers/job_queue.go b/backend/controllers/job_queue.go index 30b2fd64..f6a4116d 100644 --- a/backend/controllers/job_queue.go +++ b/backend/controllers/job_queue.go @@ -5,8 +5,9 @@ import ( ) type Job struct { - Name string - Execute func() error + Name string + ClientID string + Execute func() error } type JobQueue struct { @@ -28,28 +29,32 @@ func (q *JobQueue) AddJob(job Job) { // notify job queued go BroadcastJobStatus(JobStatus{ - Job: job.Name, - Status: "queued", + Job: job.Name, + Status: "queued", + ClientID: job.ClientID, }) } func (q *JobQueue) processJobs() { for job := range q.jobChannel { go BroadcastJobStatus(JobStatus{ - Job: job.Name, - Status: "in-progress", + Job: job.Name, + Status: "in-progress", + ClientID: job.ClientID, }) if err := job.Execute(); err != nil { go BroadcastJobStatus(JobStatus{ - Job: job.Name, - Status: "failure", + Job: job.Name, + Status: "failure", + ClientID: job.ClientID, }) } else { // utils.Logger.Infof("Success in executing job %s", job.Name) go BroadcastJobStatus(JobStatus{ - Job: job.Name, - Status: "success", + Job: job.Name, + Status: "success", + ClientID: job.ClientID, }) } diff --git a/backend/controllers/modify_task.go b/backend/controllers/modify_task.go index f0a2833d..23fcbe93 100644 --- a/backend/controllers/modify_task.go +++ b/backend/controllers/modify_task.go @@ -92,7 +92,8 @@ func ModifyTaskHandler(w http.ResponseWriter, r *http.Request) { logStore := models.GetLogStore() job := Job{ - Name: "Modify Task", + Name: "Modify Task", + ClientID: uuid, Execute: func() error { logStore.AddLog("INFO", fmt.Sprintf("Modifying task UUID: %s", taskUUID), uuid, "Modify Task") err := tw.ModifyTaskInTaskwarrior(uuid, description, project, priority, status, due, email, encryptionSecret, taskUUID, tags, depends) diff --git a/backend/controllers/websocket.go b/backend/controllers/websocket.go index 67d31b87..c7bfd42d 100644 --- a/backend/controllers/websocket.go +++ b/backend/controllers/websocket.go @@ -5,6 +5,7 @@ import ( "net/url" "os" "strings" + "sync" "ccsync_backend/utils" @@ -22,8 +23,9 @@ func getEnv() string { } type JobStatus struct { - Job string `json:"job"` - Status string `json:"status"` + Job string `json:"job"` + Status string `json:"status"` + ClientID string `json:"clientID,omitempty"` } // checkWebSocketOrigin validates the Origin header against allowed origins @@ -80,7 +82,9 @@ var upgrader = websocket.Upgrader{ CheckOrigin: checkWebSocketOrigin, } -var clients = make(map[*websocket.Conn]bool) +// clients maps each WebSocket to the authenticated user's UUID. +var clients = make(map[*websocket.Conn]string) +var clientsMu sync.RWMutex var broadcast = make(chan JobStatus) // AuthenticatedWebSocketHandler creates a WebSocket handler that requires authentication @@ -101,6 +105,15 @@ func AuthenticatedWebSocketHandler(store *sessions.CookieStore) http.HandlerFunc return } + // Prefer the session UUID over the clientID query param so a client + // cannot subscribe to another user's job-status events. + userUUID, _ := userInfo["uuid"].(string) + if userUUID == "" { + utils.Logger.Warnf("WebSocket auth failed: session user missing uuid") + http.Error(w, "Authentication required", http.StatusUnauthorized) + return + } + // User is authenticated, proceed with WebSocket upgrade ws, err := upgrader.Upgrade(w, r, nil) if err != nil { @@ -109,11 +122,18 @@ func AuthenticatedWebSocketHandler(store *sessions.CookieStore) http.HandlerFunc } defer ws.Close() - clients[ws] = true + clientsMu.Lock() + clients[ws] = userUUID + clientsMu.Unlock() + defer func() { + clientsMu.Lock() + delete(clients, ws) + clientsMu.Unlock() + }() + for { _, _, err := ws.ReadMessage() if err != nil { - delete(clients, ws) break } } @@ -130,11 +150,20 @@ func WebSocketHandler(w http.ResponseWriter, r *http.Request) { } defer ws.Close() - clients[ws] = true + // Unauthenticated legacy handler: register with an empty id so scoped + // job-status events (which always carry a ClientID) are not delivered here. + clientsMu.Lock() + clients[ws] = "" + clientsMu.Unlock() + defer func() { + clientsMu.Lock() + delete(clients, ws) + clientsMu.Unlock() + }() + for { _, _, err := ws.ReadMessage() if err != nil { - delete(clients, ws) break } } @@ -144,15 +173,26 @@ func BroadcastJobStatus(jobStatus JobStatus) { broadcast <- jobStatus } +// deliverJobStatus sends a job-status event only to WebSockets registered +// for the same ClientID as the job owner. +func deliverJobStatus(jobStatus JobStatus) { + clientsMu.Lock() + defer clientsMu.Unlock() + + for client, clientID := range clients { + if jobStatus.ClientID == "" || clientID != jobStatus.ClientID { + continue + } + if err := client.WriteJSON(jobStatus); err != nil { + client.Close() + delete(clients, client) + } + } +} + func JobStatusManager() { for { jobStatus := <-broadcast - for client := range clients { - err := client.WriteJSON(jobStatus) - if err != nil { - client.Close() - delete(clients, client) - } - } + deliverJobStatus(jobStatus) } } diff --git a/backend/controllers/websocket_test.go b/backend/controllers/websocket_test.go new file mode 100644 index 00000000..ddce9f23 --- /dev/null +++ b/backend/controllers/websocket_test.go @@ -0,0 +1,195 @@ +package controllers + +import ( + "encoding/gob" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gorilla/sessions" + "github.com/gorilla/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func resetWebSocketClients(t *testing.T) { + t.Helper() + clientsMu.Lock() + prev := clients + clients = make(map[*websocket.Conn]string) + clientsMu.Unlock() + t.Cleanup(func() { + clientsMu.Lock() + for c := range clients { + _ = c.Close() + } + clients = prev + clientsMu.Unlock() + }) +} + +func Test_deliverJobStatus_OnlyMatchingClientReceivesEvent(t *testing.T) { + resetWebSocketClients(t) + + up := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ws, err := up.Upgrade(w, r, nil) + if err != nil { + return + } + id := r.URL.Query().Get("id") + clientsMu.Lock() + clients[ws] = id + clientsMu.Unlock() + for { + if _, _, err := ws.ReadMessage(); err != nil { + clientsMu.Lock() + delete(clients, ws) + clientsMu.Unlock() + _ = ws.Close() + return + } + } + })) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + connA, _, err := websocket.DefaultDialer.Dial(wsURL+"?id=user-a", nil) + require.NoError(t, err) + defer connA.Close() + + connB, _, err := websocket.DefaultDialer.Dial(wsURL+"?id=user-b", nil) + require.NoError(t, err) + defer connB.Close() + + // Give the server goroutines time to register both clients. + require.Eventually(t, func() bool { + clientsMu.RLock() + defer clientsMu.RUnlock() + return len(clients) == 2 + }, time.Second, 10*time.Millisecond) + + deliverJobStatus(JobStatus{ + Job: "Add Task", + Status: "success", + ClientID: "user-a", + }) + + _ = connA.SetReadDeadline(time.Now().Add(2 * time.Second)) + var got JobStatus + require.NoError(t, connA.ReadJSON(&got)) + assert.Equal(t, "Add Task", got.Job) + assert.Equal(t, "success", got.Status) + assert.Equal(t, "user-a", got.ClientID) + + _ = connB.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + err = connB.ReadJSON(&got) + assert.Error(t, err, "user-b must not receive user-a's job status") +} + +func Test_AuthenticatedWebSocketHandler_UsesSessionUUIDNotQueryClientID(t *testing.T) { + resetWebSocketClients(t) + + gob.Register(map[string]interface{}{}) + store := sessions.NewCookieStore([]byte("test-session-key-32bytes-long!!")) + handler := AuthenticatedWebSocketHandler(store) + + // Establish a session cookie the same way the OAuth callback would. + seedReq := httptest.NewRequest(http.MethodGet, "/", nil) + seedRec := httptest.NewRecorder() + session, err := store.Get(seedReq, "session-name") + require.NoError(t, err) + session.Values["user"] = map[string]interface{}{ + "email": "a@example.com", + "id": "1", + "uuid": "session-uuid-a", + } + require.NoError(t, session.Save(seedReq, seedRec)) + setCookie := seedRec.Result().Header.Get("Set-Cookie") + require.NotEmpty(t, setCookie) + + server := httptest.NewServer(handler) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + header := http.Header{} + header.Add("Cookie", setCookie) + // Query claims a different user; session UUID must win. + conn, resp, err := websocket.DefaultDialer.Dial(wsURL+"?clientID=attacker-uuid", header) + require.NoError(t, err) + defer conn.Close() + require.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode) + + require.Eventually(t, func() bool { + clientsMu.RLock() + defer clientsMu.RUnlock() + for _, id := range clients { + if id == "session-uuid-a" { + return true + } + } + return false + }, time.Second, 10*time.Millisecond) + + clientsMu.RLock() + defer clientsMu.RUnlock() + for _, id := range clients { + assert.Equal(t, "session-uuid-a", id) + assert.NotEqual(t, "attacker-uuid", id) + } +} + +func Test_AddJob_IncludesClientIDInBroadcast(t *testing.T) { + resetWebSocketClients(t) + + received := make(chan JobStatus, 16) + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + for { + select { + case status := <-broadcast: + select { + case received <- status: + default: + } + case <-stop: + return + } + } + }() + defer func() { + close(stop) + <-done + }() + + q := NewJobQueue() + q.AddJob(Job{ + Name: "Add Task", + ClientID: "user-a", + Execute: func() error { return nil }, + }) + + var sawQueued, sawSuccess bool + timeout := time.After(2 * time.Second) + for !(sawQueued && sawSuccess) { + select { + case status := <-received: + // Other controller tests may also emit on the shared broadcast channel. + if status.ClientID != "user-a" || status.Job != "Add Task" { + continue + } + if status.Status == "queued" { + sawQueued = true + } + if status.Status == "success" { + sawSuccess = true + } + case <-timeout: + t.Fatal("timed out waiting for job status events with ClientID") + } + } +}