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") + } + } +}