diff --git a/Doku/api/rest.md b/Doku/api/rest.md index e14f675..db53683 100644 --- a/Doku/api/rest.md +++ b/Doku/api/rest.md @@ -13,15 +13,15 @@ Basis: `/api/v1`. Antworten sind JSON; Fehler haben die Form `{"error": "Text"}` | POST | `/auth/login` | Anmeldung (`login`/`email`/`username`, `password`) → `{token, user}` | | POST | `/auth/quickstart` | Immer `403` (Gastzugang deaktiviert) | | POST | `/webhooks/ingest` | Aufgabe per Webhook anlegen, Token als `?token=`, `X-API-Key` oder Bearer. Body: `title`/`content`/`text`/`message`, `priority`, `source`, `tags`, `dueDate`; Klartext-Body wird als Inhalt genommen | -| GET | `/ws`, `/ws/spaces/{id}`, `/ws/lists/{id}` | WebSocket, siehe [websocket.md](websocket.md) | -| GET | `/events` | SSE-Fallback | -| POST | `/events/rooms` | SSE-Räume beitreten/verlassen ([FLAW-02](../flaws.md)) | +| GET | `/ws`, `/ws/spaces/{id}`, `/ws/lists/{id}` | WebSocket, siehe [websocket.md](websocket.md); Token als `?token=` nötig | +| GET | `/events` | SSE-Fallback; Token als `?token=` nötig | ## Konto und Allgemeines (angemeldet) | Methode | Pfad | Zweck | |---|---|---| | GET | `/auth/me` | Benutzer und Kennzahlen | +| POST | `/events/rooms` | SSE-Raum beitreten/verlassen (`sseId`, `room`, `action`); nur eigene `sseId`, nur Räume mit Lesezugriff | | POST | `/auth/logout` | Token sperren | | POST | `/auth/revoke-all` | Alle Sitzungen beenden (`token_version` +1) | | PATCH | `/auth/settings` | `displayName`, `email`, `password`, `avatarUrl`, `alarmChannels`, `alarmEmail`, `discordWebhookUrl`, `customWebhookUrl`, `notifyBeforeMinutes` | diff --git a/Doku/api/websocket.md b/Doku/api/websocket.md index c2fe7e7..16a88ea 100644 --- a/Doku/api/websocket.md +++ b/Doku/api/websocket.md @@ -13,7 +13,9 @@ Der Server verteilt Änderungen als JSON-Ereignisse an verbundene Clients. Die E Der Client versucht zuerst WebSocket. Gelingt der Aufbau nicht innerhalb von 2,5 s oder bricht die Verbindung unsauber ab, wechselt er auf SSE und bleibt dabei (Firmen-Proxys). Dort liefert der Server `SSE_CONNECTED` mit einer `sseId`; Räume werden über `POST /api/v1/events/rooms` (`sseId`, `room`, `action`: `join`/`leave`) abonniert. -**Authentifizierung:** Token als `?token=` oder `Authorization`-Header. Ohne gültiges Token wird die Verbindung als `anonymous` akzeptiert ([FLAW-01](../flaws.md), [FLAW-02](../flaws.md)). Das Token wird nur beim Verbinden geprüft. +**Authentifizierung:** Token als `?token=` oder `Authorization`-Header. Ohne gültiges Token antworten `/ws` und `/events` mit `401`. Das Token wird nur beim Verbinden geprüft ([FLAW-22](../flaws.md)). + +**Räume:** Jeder Beitritt (Raum in der URL, `JOIN_ROOM`, SSE-Startparameter `rooms`/`docId`, `POST /events/rooms`) wird geprüft: `doc:` nur mit Lesezugriff auf die Notiz, sonst muss die ID eine lesbare geteilte Liste sein. Nicht erlaubte Räume werden ignoriert, `POST /events/rooms` antwortet dann mit `403`. SSE-Räume kann nur der Besitzer der jeweiligen `sseId` ändern. ## Nachrichtenformat @@ -29,20 +31,20 @@ Client → Server versteht der Server nur: `{"event":"JOIN_ROOM","room":""}` |---|---| | Benutzer-Kanal (`BroadcastToUser`) | alle Verbindungen eines Benutzers | | Raum (`BroadcastRoom`, `Broadcast`) | Verbindungen im Raum | -| Alle (`BroadcastToAll`) | **jede** Verbindung, auch unangemeldete | +| Mehrere Benutzer (`BroadcastToUsers`) | alle Verbindungen einer berechneten Benutzergruppe, z. B. Eigentümer und Mitglieder einer Aufgabenliste | ## Ereignisse | Ereignis | Auslöser | Kanal | |---|---|---| -| `TASK_CREATED` | `POST /inbox` | Benutzer | -| `TASK_UPDATED` | `PATCH /thoughts/{id}` (Triage: Benutzer) | alle | -| `TASK_DELETED` | `DELETE /thoughts/{id}` | alle | -| `TASK_REORDERED` | Aufgaben umsortieren | alle | -| `TASK_LIST_UPDATED` / `TASK_LIST_DELETED` | Liste ändern, (un)teilen, löschen | alle | +| `TASK_CREATED` | `POST /inbox`, Webhook | Eigentümer und Benutzer der Liste | +| `TASK_UPDATED` | `PATCH /thoughts/{id}`, Triage | Eigentümer und Benutzer der alten und neuen Liste | +| `TASK_DELETED` | `DELETE /thoughts/{id}` | Eigentümer und Benutzer der Liste | +| `TASK_REORDERED` | Aufgaben umsortieren | Benutzer aller betroffenen Listen | +| `TASK_LIST_UPDATED` / `TASK_LIST_DELETED` | Liste ändern, (un)teilen, löschen; Payload nur `{id}`, Clients laden ihre Listen neu | Benutzer vor und nach der Änderung | | `DOC_CREATED`, `DOC_UPDATED`, `DOC_DELETED` | Notiz anlegen/ändern/löschen, Freigaben | berechtigte Benutzer | | `DOC_BLOCKS_PATCHED` | `PATCH /documents/{id}/blocks` (`docId`, `updatedBy`, `upserted`, `deletedIds`, `blockOrder`) | Raum `doc:` und berechtigte Benutzer | -| `WORKSPACE_FOLDERS_UPDATED` | Ordner/Mounts ändern | Benutzer | +| `WORKSPACE_FOLDERS_UPDATED` | Ordner/Mounts ändern | Benutzer selbst; bei geteilten Ordnern Eigentümer und alle Beteiligten | | `LIST_CREATED`, `LIST_UPDATED`, `LIST_DELETED`, `LIST_SHARED`, `LIST_UNSHARED`, `LIST_CLEARED` | Listenverwaltung | Benutzer bzw. Raum der Liste | | `ITEM_CREATED`, `ITEM_UPDATED`, `ITEM_TOGGLED`, `ITEM_DELETED`, `ITEM_REORDERED` | Listeneinträge | Raum der Liste | | `USER_PRESENCE` | Verbindung auf-/abgebaut (`onlineCount`) | Raum | diff --git a/Doku/flaws.md b/Doku/flaws.md index b7a4112..7a6043d 100644 --- a/Doku/flaws.md +++ b/Doku/flaws.md @@ -10,9 +10,9 @@ Schweregrade: **Hoch** (Datenabfluss oder Rechteumgehung möglich), **Mittel** ( | ID | Bereich | Kurzbeschreibung | Schwere | |---|---|---|---| -| FLAW-01 | Realtime | `BroadcastToAll` sendet Aufgaben- und Listendaten an alle Verbindungen, auch unauthentifizierte | **Hoch** | -| FLAW-02 | Realtime | Räume (WS `JOIN_ROOM`, SSE `/events/rooms`, `?listId=`) ohne Berechtigungsprüfung | **Hoch** | -| FLAW-03 | Aufgaben | `ReorderTasks` ändert beliebige Aufgaben per ID, ohne Zugriffsprüfung (IDOR) | **Hoch** | +| FLAW-01 | Realtime | ~~`BroadcastToAll` sendet Aufgaben- und Listendaten an alle Verbindungen~~ | **Behoben** | +| FLAW-02 | Realtime | ~~Räume ohne Berechtigungsprüfung~~ | **Behoben** | +| FLAW-03 | Aufgaben | ~~`ReorderTasks` ändert beliebige Aufgaben per ID (IDOR)~~ | **Behoben** | | FLAW-04 | Auth | Fest eincodiertes Fallback-`JWT_SECRET` | **Hoch** | | FLAW-05 | Benachrichtigungen | Discord-/Webhook-URLs ungeprüft (SSRF) | Mittel | | FLAW-06 | Aufgaben | Jeder Editor einer Aufgabenliste kann sie per `teamId` an weitere Gruppen freigeben | Mittel | @@ -24,7 +24,7 @@ Schweregrade: **Hoch** (Datenabfluss oder Rechteumgehung möglich), **Mittel** ( | FLAW-12 | Benachrichtigungen | `notify_before_minutes` wirkungslos; Alarm feuert nach Umplanung nie wieder | Mittel | | FLAW-13 | Datenbank | Keine Schema-Versionierung; Migrationsfehler werden verschluckt; `migrations/001_init.sql` ist toter Code | Mittel | | FLAW-14 | Backend | Fehler von Schreiboperationen werden häufig ignoriert (`_, _ =`) | Mittel | -| FLAW-15 | Realtime | `WriteTimeout` von 30 s trifft vermutlich auch SSE-Streams *(nicht live verifiziert)* | Mittel | +| FLAW-15 | Realtime | ~~`WriteTimeout` von 30 s beendet SSE-Streams~~ | **Behoben** | | FLAW-16 | Setup | `/auth/setup` ist bis zum ersten Konto offen; Demo-Modus legt `admin/admin` an | Mittel | | FLAW-17 | Workspace | Dokument- und Ordnerfreigaben sind per Migration erzwungen schreibgeschützt; tote Schreibpfade | Niedrig | | FLAW-18 | Suche | Nur eigene Gedanken und Dokumente, `LIKE` ohne Escaping, keine FTS | Niedrig | @@ -40,16 +40,19 @@ Schweregrade: **Hoch** (Datenabfluss oder Rechteumgehung möglich), **Mittel** ( ## Details ### FLAW-01: Broadcast an alle Verbindungen (Datenabfluss) +- **Status: Behoben.** `BroadcastToAll` existiert nicht mehr. Aufgabenereignisse gehen an Eigentümer und Benutzer der Liste (`GetTaskListUserIDs`), Listenereignisse tragen nur noch die ID, Ordnerereignisse gehen an Benutzer bzw. Ordnerbeteiligte (`GetSharedFolderUserIDs`). Test: `TestTaskEventsOnlyReachAuthorizedUsers`. - **Fundstellen:** [`internal/ws/hub.go`](../internal/ws/hub.go) `BroadcastToAll`; Aufrufer in [`internal/api/server.go`](../internal/api/server.go): `handleUpdateThought` (`TASK_UPDATED`), `handleDeleteThought` (`TASK_DELETED`), `handleReorderTasks` (`TASK_REORDERED`), `handleUpdateTaskList`, `handleCreateTaskListShare`, `handleDeleteTaskListShare` (`TASK_LIST_UPDATED`), `handleDeleteTaskList`. - **Problem:** `BroadcastToAll` schreibt an `allClients` und `sseClients`. Dazu zählen auch Verbindungen mit `userID = "anonymous"`, denn `handleWebSocket` und `handleSSE` liegen außerhalb der Auth-Gruppe und lassen jede Verbindung zu. Der vollständige Aufgabeninhalt (Titel, Notizen, Unteraufgaben) und die Listen samt Mitgliedern gehen so an jeden, der `/api/v1/events` öffnet. - **Maßnahme:** Ereignisse an die berechtigten Nutzer adressieren (wie bei `DOC_*` über `GetAuthorizedDocUsers`), unauthentifizierte Verbindungen ablehnen. ### FLAW-02: Räume ohne Berechtigungsprüfung +- **Status: Behoben.** `/ws` und `/events` verlangen ein gültiges Token (sonst `401`). Jeder Raumbeitritt läuft über `Hub.SetRoomAuthorizer` → `Server.canJoinRoom` (Notiz bzw. Liste muss lesbar sein). `/events/rooms` liegt hinter der Anmeldung und akzeptiert nur die eigene `sseId`. Tests: `TestRealtimeEndpointsRequireAuth`, `TestSSERoomJoinRequiresAccess`, `TestRoomJoinsRequireAuthorization`. - **Fundstellen:** `Hub.readPump` (`JOIN_ROOM`), `handleSSERooms` (Route `/api/v1/events/rooms` liegt **außerhalb** der Auth-Gruppe), `Hub.Register` (Raum aus `?listId=`/`?spaceId=`). - **Problem:** Wer eine Raum-ID kennt (`doc:`, Listen-ID), tritt bei und erhält Inhalte wie `DOC_BLOCKS_PATCHED` oder `ITEM_*`. Raum-IDs tauchen in Ereignissen auf und sind nicht geheim. Für den SSE-Beitritt reicht die `sseId`, die der Server selbst im Stream mitsendet. - **Maßnahme:** Beitritt nur nach Zugriffsprüfung (`GetDocumentByID`, `GetListByID`); `/events/rooms` hinter `authMiddleware`. ### FLAW-03: `ReorderTasks` ohne Zugriffsprüfung (IDOR) +- **Status: Behoben.** `handleReorderTasks` lehnt die ganze Anfrage ab, sobald eine Aufgabe nicht zugreifbar (`404`) oder schreibgeschützt (`403`) ist. Test: `TestReorderRejectsForeignTasks`. - **Fundstelle:** `Store.ReorderTasks` (`UPDATE thoughts SET sort_order … WHERE id = ?`) und `handleReorderTasks`. - **Problem:** Die SQL-Anweisung ist weder auf `user_id` noch auf die Liste eingeschränkt. Jeder angemeldete Nutzer kann `sortOrder`, `kanbanStatus` und damit `isCompleted` fremder Aufgaben ändern, sofern er die ID kennt. Die Leser-Prüfung für Aufgabenlisten überspringt Aufgaben, auf die der Nutzer gar keinen Zugriff hat (`GetThoughtByID` liefert dann einen Fehler), und schützt davor nicht. - **Maßnahme:** Jede ID gegen `accessibleTaskListIDsQuery` bzw. `user_id` prüfen und bei Verstoß ablehnen. @@ -119,7 +122,8 @@ Schweregrade: **Hoch** (Datenabfluss oder Rechteumgehung möglich), **Mittel** ( - **Problem:** Scheitert ein Schreibzugriff (z. B. `SQLITE_BUSY`), antwortet die API trotzdem mit Erfolg und broadcastet den Stand, der nie gespeichert wurde. Fehler lassen sich nur schwer nachvollziehen. - **Maßnahme:** Mehrstufige Änderungen in Transaktionen bündeln und Fehler zurückgeben. -### FLAW-15: WriteTimeout und SSE *(nicht live verifiziert)* +### FLAW-15: WriteTimeout und SSE +- **Status: Behoben.** `ServeSSE` hebt die Schreibfrist für die eigene Antwort auf (`http.ResponseController.SetWriteDeadline`). Test mit 1-s-Timeout: `TestSSEStreamOutlivesWriteTimeout`. Der Test hat den Abbruch im alten Code bestätigt. - **Fundstelle:** [`cmd/server/main.go`](../cmd/server/main.go) `WriteTimeout: 30 * time.Second`. - **Problem:** Go beendet nach diesem Zeitraum das Schreiben der Antwort. Für den SSE-Fallback (`ServeSSE`) bedeutet das voraussichtlich einen Verbindungsabbruch alle 30 s. WebSockets sind durch Hijacking nicht betroffen. Der Client verbindet sich neu, daher fällt es kaum auf. - **Maßnahme:** Im SSE-Handler `http.ResponseController.SetWriteDeadline(time.Time{})` setzen oder `WriteTimeout` für diese Route aufheben. diff --git a/Doku/plans/kollaboration-notizen.md b/Doku/plans/kollaboration-notizen.md index 0ce21ed..a1ee83e 100644 --- a/Doku/plans/kollaboration-notizen.md +++ b/Doku/plans/kollaboration-notizen.md @@ -277,8 +277,8 @@ Damit private Notizen keinen unnötigen Aufwand erzeugen, liefert der Server im | Phase | Inhalt | Ergebnis | |---|---|---| -| 0 | Sicherheitsfixes vorziehen: Raumberechtigung (FLAW-01/02), Stream-Timeout (FLAW-15), zentrale `docRole` | Grundlage dicht | -| 1 | Neues Schema, Operations-Log, `/snapshot`, `/sync`, Konfliktregeln, Simulationstests | Server fertig und abgesichert | +| 0 | Sicherheitsfixes vorziehen: Raumberechtigung (FLAW-01/02), Umsortieren (FLAW-03), Stream-Timeout (FLAW-15). **Erledigt.** Die zentrale `docRole` wandert in Phase 1, weil sie auf dem neuen Freigabemodell (`workspace_access`) aufbaut. | Grundlage dicht | +| 1 | Neues Schema inkl. `workspace_access` und zentraler `docRole`, Operations-Log, `/snapshot`, `/sync`, Konfliktregeln, Simulationstests | Server fertig und abgesichert | | 2 | `SyncEngine` mit IndexedDB-Warteschlange, nur Intervall-Abfrage als Transport, Editor umgestellt (Baum statt `level`) | Bearbeiten funktioniert überall, auch offline | | 3 | Leases und Präsenz in UI, Konfliktblöcke mit Übernehmen/Verwerfen, Rückgängig/Wiederholen | Gemeinsames Bearbeiten ohne Kollisionen, Strg+Z | | 4 | Push-Stufen WebSocket → SSE → Long-Polling mit automatischer Erkennung | Live-Gefühl, wo das Netz es zulässt | diff --git a/frontend/src/features/triage/TasksView.tsx b/frontend/src/features/triage/TasksView.tsx index 5dd7628..2ee010e 100644 --- a/frontend/src/features/triage/TasksView.tsx +++ b/frontend/src/features/triage/TasksView.tsx @@ -220,6 +220,9 @@ export const TasksView: React.FC = ({ } } else if (msg.event === 'TASK_REORDERED') { loadData(); + } else if (msg.event === 'TASK_LIST_UPDATED' || msg.event === 'TASK_LIST_DELETED') { + // Payload only carries the list ID; reload so permissions reflect this user's view + api.getTaskLists().then((lists) => setTaskLists(Array.isArray(lists) ? lists : [])).catch(() => {}); } }; diff --git a/internal/api/realtime_test.go b/internal/api/realtime_test.go new file mode 100644 index 0000000..51520a0 --- /dev/null +++ b/internal/api/realtime_test.go @@ -0,0 +1,268 @@ +package api + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/sandrodach/mindsift/internal/auth" + "github.com/sandrodach/mindsift/internal/domain" + "github.com/sandrodach/mindsift/internal/storage" +) + +const testJWTSecret = "test-jwt-secret-key" + +type sseMsg struct { + Event string `json:"event"` + Payload json.RawMessage `json:"payload"` +} + +type sseStream struct { + id string + events chan sseMsg +} + +// startRealtimeServer serves the router over real HTTP with a short WriteTimeout, like production but faster. +func startRealtimeServer(t *testing.T) (*Server, *storage.Store, *httptest.Server, *domain.User, string) { + t.Helper() + srv, store, token := setupTestServer(t) + claims, err := auth.ValidateToken(testJWTSecret, token) + if err != nil { + t.Fatalf("token: %v", err) + } + owner, _ := store.GetUserByID(claims.UserID) + ts := httptest.NewUnstartedServer(srv.Router()) + ts.Config.WriteTimeout = time.Second + ts.Start() + t.Cleanup(ts.Close) + return srv, store, ts, owner, token +} + +func newTestUser(t *testing.T, store *storage.Store, email string) (*domain.User, string) { + t.Helper() + u, err := store.CreateUser(email, "hash", strings.Split(email, "@")[0], "member") + if err != nil { + t.Fatalf("create user: %v", err) + } + tok, err := auth.GenerateToken(testJWTSecret, u.ID, u.Email, u.DisplayName, u.TokenVersion) + if err != nil { + t.Fatalf("token: %v", err) + } + return u, tok +} + +func openSSE(t *testing.T, ts *httptest.Server, token string) *sseStream { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, ts.URL+"/api/v1/events?token="+token, nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + cancel() + t.Fatalf("open sse: %v", err) + } + if resp.StatusCode != http.StatusOK { + cancel() + t.Fatalf("open sse: status %d", resp.StatusCode) + } + st := &sseStream{events: make(chan sseMsg, 64)} + go func() { + defer close(st.events) + sc := bufio.NewScanner(resp.Body) + for sc.Scan() { + if line, ok := strings.CutPrefix(sc.Text(), "data: "); ok { + var m sseMsg + if json.Unmarshal([]byte(line), &m) == nil { + st.events <- m + } + } + } + }() + t.Cleanup(func() { + cancel() + _ = resp.Body.Close() + }) + connected := st.waitFor(t, "SSE_CONNECTED", 2*time.Second) + if connected == nil { + t.Fatal("no SSE_CONNECTED event") + } + var p struct { + SSEID string `json:"sseId"` + } + _ = json.Unmarshal(connected.Payload, &p) + st.id = p.SSEID + return st +} + +func (st *sseStream) waitFor(t *testing.T, event string, timeout time.Duration) *sseMsg { + t.Helper() + deadline := time.After(timeout) + for { + select { + case m, ok := <-st.events: + if !ok { + return nil + } + if m.Event == event { + return &m + } + case <-deadline: + return nil + } + } +} + +func doJSON(t *testing.T, ts *httptest.Server, method, path, token string, body interface{}) *http.Response { + t.Helper() + var buf bytes.Buffer + if body != nil { + _ = json.NewEncoder(&buf).Encode(body) + } + req, _ := http.NewRequest(method, ts.URL+path, &buf) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("%s %s: %v", method, path, err) + } + t.Cleanup(func() { _ = resp.Body.Close() }) + return resp +} + +func TestRealtimeEndpointsRequireAuth(t *testing.T) { + _, _, ts, _, _ := startRealtimeServer(t) + for _, path := range []string{"/api/v1/events", "/api/v1/ws", "/api/v1/events/rooms"} { + method := http.MethodGet + if strings.HasSuffix(path, "/rooms") { + method = http.MethodPost + } + req, _ := http.NewRequest(method, ts.URL+path, strings.NewReader(`{}`)) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("%s: %v", path, err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Errorf("%s without token: want 401, got %d", path, resp.StatusCode) + } + } +} + +func TestSSEStreamOutlivesWriteTimeout(t *testing.T) { + srv, _, ts, owner, token := startRealtimeServer(t) + st := openSSE(t, ts, token) + + time.Sleep(1500 * time.Millisecond) // longer than the server's WriteTimeout of 1s + srv.wsHub.BroadcastToUser(owner.ID, "LATE_EVENT", map[string]string{"ok": "1"}) + + if st.waitFor(t, "LATE_EVENT", 2*time.Second) == nil { + t.Fatal("SSE stream was cut by the WriteTimeout") + } +} + +func TestTaskEventsOnlyReachAuthorizedUsers(t *testing.T) { + _, store, ts, owner, ownerToken := startRealtimeServer(t) + member, memberToken := newTestUser(t, store, "member@example.com") + _, outsiderToken := newTestUser(t, store, "outsider@example.com") + + list, err := store.CreateTaskList(owner.ID, "", "", "Haushalt", "🏠", "#000000") + if err != nil { + t.Fatalf("create list: %v", err) + } + if _, err := store.AddTaskListShare(list.ID, "user", member.ID, "viewer", owner.ID); err != nil { + t.Fatalf("share: %v", err) + } + task, err := store.CreateThoughtWithDetails(owner.ID, "Fenster putzen", "normal", "manual", "", list.ID, "todo", "") + if err != nil { + t.Fatalf("create task: %v", err) + } + + memberStream := openSSE(t, ts, memberToken) + outsiderStream := openSSE(t, ts, outsiderToken) + + resp := doJSON(t, ts, http.MethodPatch, "/api/v1/thoughts/"+task.ID, ownerToken, map[string]string{"content": "Fenster putzen (geheim)"}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("update task: status %d", resp.StatusCode) + } + if memberStream.waitFor(t, "TASK_UPDATED", 2*time.Second) == nil { + t.Error("list member did not receive TASK_UPDATED") + } + if outsiderStream.waitFor(t, "TASK_UPDATED", 500*time.Millisecond) != nil { + t.Error("outsider received TASK_UPDATED of a foreign list") + } + + resp = doJSON(t, ts, http.MethodPut, "/api/v1/task-lists/"+list.ID, ownerToken, map[string]string{"title": "Haushalt 2"}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("update list: status %d", resp.StatusCode) + } + msg := memberStream.waitFor(t, "TASK_LIST_UPDATED", 2*time.Second) + if msg == nil { + t.Fatal("list member did not receive TASK_LIST_UPDATED") + } + if strings.Contains(string(msg.Payload), "permission") || !strings.Contains(string(msg.Payload), list.ID) { + t.Errorf("TASK_LIST_UPDATED must only carry the list ID, got %s", msg.Payload) + } + if outsiderStream.waitFor(t, "TASK_LIST_UPDATED", 500*time.Millisecond) != nil { + t.Error("outsider received TASK_LIST_UPDATED of a foreign list") + } +} + +func TestSSERoomJoinRequiresAccess(t *testing.T) { + _, store, ts, owner, _ := startRealtimeServer(t) + friend, friendToken := newTestUser(t, store, "friend@example.com") + _, strangerToken := newTestUser(t, store, "stranger@example.com") + + list, err := store.CreateList(owner.ID, "Einkauf", "🛒", "shopping") + if err != nil { + t.Fatalf("create list: %v", err) + } + friendStream := openSSE(t, ts, friendToken) + strangerStream := openSSE(t, ts, strangerToken) + + join := func(token, sseID string) int { + resp := doJSON(t, ts, http.MethodPost, "/api/v1/events/rooms", token, map[string]string{"sseId": sseID, "room": list.ID}) + return resp.StatusCode + } + + if code := join(strangerToken, strangerStream.id); code != http.StatusForbidden { + t.Errorf("stranger joining a foreign list room: want 403, got %d", code) + } + if code := join(friendToken, friendStream.id); code != http.StatusForbidden { + t.Errorf("friend joining before being shared: want 403, got %d", code) + } + if _, err := store.AddListShare(owner.ID, list.ID, "user", friend.ID, "viewer"); err != nil { + t.Fatalf("share list: %v", err) + } + if code := join(strangerToken, friendStream.id); code != http.StatusForbidden { + t.Errorf("joining with someone else's sseId: want 403, got %d", code) + } + if code := join(friendToken, friendStream.id); code != http.StatusOK { + t.Errorf("friend joining shared list room: want 200, got %d", code) + } +} + +func TestReorderRejectsForeignTasks(t *testing.T) { + _, store, ts, owner, _ := startRealtimeServer(t) + _, attackerToken := newTestUser(t, store, "attacker@example.com") + + task, err := store.CreateThoughtWithDetails(owner.ID, "Privat", "normal", "manual", "", "", "todo", "") + if err != nil { + t.Fatalf("create task: %v", err) + } + done := "done" + resp := doJSON(t, ts, http.MethodPost, "/api/v1/tasks/reorder", attackerToken, map[string]interface{}{ + "items": []domain.ThoughtReorderItem{{ID: task.ID, SortOrder: 99, KanbanStatus: &done}}, + }) + if resp.StatusCode == http.StatusOK { + t.Fatal("reordering a foreign task must be rejected") + } + after, _ := store.GetThoughtByID(owner.ID, task.ID) + if after.SortOrder == 99 || after.IsCompleted { + t.Fatalf("foreign task was modified: sortOrder=%d completed=%v", after.SortOrder, after.IsCompleted) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index 0fcc16b..0b9c0dc 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -38,10 +38,29 @@ func NewServer(cfg *config.Config, store *storage.Store, wsHub *ws.Hub, staticFS wsHub: wsHub, staticFS: staticFS, } + wsHub.SetRoomAuthorizer(srv.canJoinRoom) go srv.startAlarmTicker() return srv } +// canJoinRoom allows realtime rooms only to users with read access: "doc:" for notes, otherwise a shared list ID. +func (s *Server) canJoinRoom(userID, room string) bool { + if docID, ok := strings.CutPrefix(room, "doc:"); ok { + _, err := s.store.GetDocumentByID(userID, docID) + return err == nil + } + _, err := s.store.GetListByID(userID, room) + return err == nil +} + +// taskAudience returns everyone who can see a task: its owner plus all users of its task list. +func (s *Server) taskAudience(th *domain.Thought) []string { + if th == nil { + return nil + } + return append(s.store.GetTaskListUserIDs(th.TaskListID), th.UserID) +} + func (s *Server) Router() http.Handler { r := chi.NewRouter() @@ -69,13 +88,13 @@ func (s *Server) Router() http.Handler { r.Get("/api/v1/ws/spaces/{spaceId}", s.handleWebSocket) r.Get("/api/v1/ws/lists/{listId}", s.handleWebSocket) r.Get("/api/v1/events", s.handleSSE) - r.Post("/api/v1/events/rooms", s.handleSSERooms) // Authenticated API Routes r.Group(func(r chi.Router) { r.Use(s.authMiddleware) r.Get("/api/v1/auth/me", s.handleMe) + r.Post("/api/v1/events/rooms", s.handleSSERooms) r.Post("/api/v1/auth/logout", s.handleLogout) r.Post("/api/v1/auth/revoke-all", s.handleRevokeAllSessions) r.Patch("/api/v1/auth/settings", s.handleUpdateUserSettings) @@ -699,7 +718,7 @@ func (s *Server) handleCaptureInbox(w http.ResponseWriter, r *http.Request) { } } - s.wsHub.BroadcastTaskEvent(thought.UserID, "TASK_CREATED", thought) + s.wsHub.BroadcastToUsers(s.taskAudience(thought), "TASK_CREATED", thought) writeJSON(w, http.StatusCreated, thought) } @@ -785,7 +804,7 @@ func (s *Server) handleTriageAction(w http.ResponseWriter, r *http.Request) { } } - s.wsHub.BroadcastTaskEvent(updated.UserID, "TASK_UPDATED", updated) + s.wsHub.BroadcastToUsers(s.taskAudience(updated), "TASK_UPDATED", updated) writeJSON(w, http.StatusOK, updated) } @@ -819,11 +838,19 @@ func (s *Server) handleReorderTasks(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusForbidden, storage.ErrTaskListReadOnly.Error()) return } + // ReorderTasks updates by ID only, so every item must be a task the caller may edit + audience := []string{userID} for _, it := range req.Items { - if th, err := s.store.GetThoughtByID(userID, it.ID); err == nil && th.TaskListID != "" && !s.store.CanEditTaskList(userID, th.TaskListID) { + th, err := s.store.GetThoughtByID(userID, it.ID) + if err != nil { + writeError(w, http.StatusNotFound, "Aufgabe nicht gefunden") + return + } + if th.TaskListID != "" && !s.store.CanEditTaskList(userID, th.TaskListID) { writeError(w, http.StatusForbidden, storage.ErrTaskListReadOnly.Error()) return } + audience = append(audience, s.taskAudience(th)...) } if err := s.store.ReorderTasks(taskListID, req.Items); err != nil { @@ -831,7 +858,7 @@ func (s *Server) handleReorderTasks(w http.ResponseWriter, r *http.Request) { return } - s.wsHub.BroadcastToAll("TASK_REORDERED", map[string]interface{}{ + s.wsHub.BroadcastToUsers(audience, "TASK_REORDERED", map[string]interface{}{ "taskListId": taskListID, "items": req.Items, "userId": userID, @@ -859,13 +886,14 @@ func (s *Server) handleUpdateThought(w http.ResponseWriter, r *http.Request) { return } + before, _ := s.store.GetThoughtByID(userID, thoughtID) updated, err := s.store.UpdateThoughtDetails(userID, thoughtID, req.Content, req.IsCompleted, req.DueDate, req.Priority, req.TaskListID, req.KanbanStatus, req.Notes, req.Subtasks, req.Tags) if err != nil { writeStoreError(w, http.StatusBadRequest, err) return } s.store.SyncDocBlockAndTask(userID, thoughtID, updated.IsCompleted) - s.wsHub.BroadcastToAll("TASK_UPDATED", updated) + s.wsHub.BroadcastToUsers(append(s.taskAudience(before), s.taskAudience(updated)...), "TASK_UPDATED", updated) writeJSON(w, http.StatusOK, updated) } @@ -879,16 +907,18 @@ func (s *Server) handleUnlinkEnhancedTask(w http.ResponseWriter, r *http.Request func (s *Server) handleDeleteThought(w http.ResponseWriter, r *http.Request) { userID := auth.GetUserID(r.Context()) thoughtID := chi.URLParam(r, "id") - if th, err := s.store.GetThoughtByID(userID, thoughtID); err == nil && th.TaskListID != "" && !s.store.CanEditTaskList(userID, th.TaskListID) { + th, err := s.store.GetThoughtByID(userID, thoughtID) + if err == nil && th.TaskListID != "" && !s.store.CanEditTaskList(userID, th.TaskListID) { writeError(w, http.StatusForbidden, storage.ErrTaskListReadOnly.Error()) return } + audience := s.taskAudience(th) s.store.UnlinkTaskFromDocBlock(userID, thoughtID) if err := s.store.DeleteThought(userID, thoughtID); err != nil { writeStoreError(w, http.StatusInternalServerError, err) return } - s.wsHub.BroadcastToAll("TASK_DELETED", map[string]string{"id": thoughtID}) + s.wsHub.BroadcastToUsers(audience, "TASK_DELETED", map[string]string{"id": thoughtID}) writeJSON(w, http.StatusOK, map[string]bool{"deleted": true}) } @@ -1096,7 +1126,7 @@ func (s *Server) handleCreateFolder(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, err.Error()) return } - s.wsHub.BroadcastDocEvent("", "WORKSPACE_FOLDERS_UPDATED", map[string]string{"path": cleanPath}) + s.wsHub.BroadcastToUser(userID, "WORKSPACE_FOLDERS_UPDATED", map[string]string{"path": cleanPath}) writeJSON(w, http.StatusCreated, map[string]string{"path": cleanPath}) } @@ -1117,7 +1147,7 @@ func (s *Server) handleDeleteFolder(w http.ResponseWriter, r *http.Request) { for _, docID := range deletedDocIDs { s.wsHub.BroadcastDocEvent(docID, "DOC_DELETED", map[string]string{"id": docID}) } - s.wsHub.BroadcastDocEvent("", "WORKSPACE_FOLDERS_UPDATED", map[string]string{"path": folderPath}) + s.wsHub.BroadcastToUser(userID, "WORKSPACE_FOLDERS_UPDATED", map[string]string{"path": folderPath}) writeJSON(w, http.StatusOK, map[string]interface{}{ "deleted": true, @@ -1146,7 +1176,7 @@ func (s *Server) handleMoveFolder(w http.ResponseWriter, r *http.Request) { s.broadcastDocToAuthorizedUsers(id, "DOC_UPDATED") } } - s.wsHub.BroadcastDocEvent("", "WORKSPACE_FOLDERS_UPDATED", map[string]string{ + s.wsHub.BroadcastToUser(userID, "WORKSPACE_FOLDERS_UPDATED", map[string]string{ "sourcePath": req.SourcePath, "newPath": newPath, }) @@ -1183,7 +1213,7 @@ func (s *Server) handleCreateSharedFolder(w http.ResponseWriter, r *http.Request writeError(w, http.StatusBadRequest, err.Error()) return } - s.wsHub.BroadcastDocEvent("", "WORKSPACE_FOLDERS_UPDATED", folder) + s.wsHub.BroadcastToUser(userID, "WORKSPACE_FOLDERS_UPDATED", folder) writeJSON(w, http.StatusCreated, folder) } @@ -1201,11 +1231,12 @@ func (s *Server) handleGetSharedFolderByID(w http.ResponseWriter, r *http.Reques func (s *Server) handleDeleteSharedFolder(w http.ResponseWriter, r *http.Request) { userID := auth.GetUserID(r.Context()) folderID := chi.URLParam(r, "id") + audience := s.store.GetSharedFolderUserIDs(folderID) if err := s.store.DeleteSharedFolder(userID, folderID); err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } - s.wsHub.BroadcastDocEvent("", "WORKSPACE_FOLDERS_UPDATED", map[string]string{"folderId": folderID}) + s.wsHub.BroadcastToUsers(append(audience, userID), "WORKSPACE_FOLDERS_UPDATED", map[string]string{"folderId": folderID}) writeJSON(w, http.StatusOK, map[string]bool{"deleted": true}) } @@ -1273,7 +1304,7 @@ func (s *Server) handleCreateFolderShare(w http.ResponseWriter, r *http.Request) writeError(w, http.StatusBadRequest, err.Error()) return } - s.wsHub.BroadcastDocEvent("", "WORKSPACE_FOLDERS_UPDATED", share) + s.wsHub.BroadcastToUsers(s.store.GetSharedFolderUserIDs(folderID), "WORKSPACE_FOLDERS_UPDATED", map[string]string{"folderId": folderID}) if docIDs, err := s.store.GetSharedFolderDocIDs(folderID); err == nil { for _, docID := range docIDs { s.broadcastDocToAuthorizedUsers(docID, "DOC_CREATED") @@ -1287,13 +1318,24 @@ func (s *Server) handleDeleteFolderShare(w http.ResponseWriter, r *http.Request) folderID := chi.URLParam(r, "id") shareID := chi.URLParam(r, "shareId") docIDs, _ := s.store.GetSharedFolderDocIDs(folderID) + audienceBefore := s.store.GetSharedFolderUserIDs(folderID) if err := s.store.DeleteFolderShare(userID, folderID, shareID); err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } - s.wsHub.BroadcastDocEvent("", "WORKSPACE_FOLDERS_UPDATED", map[string]string{"folderId": folderID, "shareId": shareID}) + audienceAfter := s.store.GetSharedFolderUserIDs(folderID) + stillAllowed := make(map[string]bool, len(audienceAfter)) + for _, uid := range audienceAfter { + stillAllowed[uid] = true + } + s.wsHub.BroadcastToUsers(append(audienceBefore, audienceAfter...), "WORKSPACE_FOLDERS_UPDATED", map[string]string{"folderId": folderID}) for _, docID := range docIDs { - s.wsHub.BroadcastToUser(shareID, "DOC_DELETED", map[string]string{"id": docID}) + // Users who lost access remove the notes; the rest refresh them + for _, uid := range audienceBefore { + if !stillAllowed[uid] { + s.wsHub.BroadcastToUser(uid, "DOC_DELETED", map[string]string{"id": docID}) + } + } s.broadcastDocToAuthorizedUsers(docID, "DOC_UPDATED") } writeJSON(w, http.StatusOK, map[string]bool{"deleted": true}) @@ -1316,7 +1358,8 @@ func (s *Server) handleMoveDocument(w http.ResponseWriter, r *http.Request) { } s.wsHub.BroadcastDocEvent(docID, "DOC_UPDATED", doc) s.broadcastDocToAuthorizedUsers(docID, "DOC_UPDATED") - s.wsHub.BroadcastDocEvent("", "WORKSPACE_FOLDERS_UPDATED", nil) + docUsers, _ := s.store.GetAuthorizedDocUsers(docID) + s.wsHub.BroadcastToUsers(append(docUsers, userID), "WORKSPACE_FOLDERS_UPDATED", map[string]string{"docId": docID}) writeJSON(w, http.StatusOK, doc) } @@ -1751,7 +1794,8 @@ func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) { userID := s.authenticateUserFromReq(r) if userID == "" { - userID = "anonymous" + writeError(w, http.StatusUnauthorized, "Nicht autorisiert – bitte anmelden") + return } s.wsHub.ServeWS(w, r, roomID, userID) } @@ -1764,7 +1808,8 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { userID := s.authenticateUserFromReq(r) if userID == "" { - userID = "anonymous" + writeError(w, http.StatusUnauthorized, "Nicht autorisiert – bitte anmelden") + return } var initialRooms []string @@ -1784,6 +1829,7 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleSSERooms(w http.ResponseWriter, r *http.Request) { + userID := auth.GetUserID(r.Context()) var req struct { SSEID string `json:"sseId"` Room string `json:"room"` @@ -1803,9 +1849,13 @@ func (s *Server) handleSSERooms(w http.ResponseWriter, r *http.Request) { } if req.Action == "leave" { - s.wsHub.LeaveSSERoomByID(req.SSEID, req.Room) - } else { - s.wsHub.JoinSSERoomByID(req.SSEID, req.Room) + s.wsHub.LeaveSSERoomByID(req.SSEID, userID, req.Room) + writeJSON(w, http.StatusOK, map[string]bool{"success": true}) + return + } + if !s.wsHub.JoinSSERoomByID(req.SSEID, userID, req.Room) { + writeError(w, http.StatusForbidden, "Kein Zugriff auf diesen Raum") + return } writeJSON(w, http.StatusOK, map[string]bool{"success": true}) } @@ -1943,7 +1993,7 @@ func (s *Server) handleWebhookIngest(w http.ResponseWriter, r *http.Request) { return } - s.wsHub.BroadcastTaskEvent(thought.UserID, "TASK_CREATED", thought) + s.wsHub.BroadcastToUsers(s.taskAudience(thought), "TASK_CREATED", thought) writeJSON(w, http.StatusCreated, map[string]interface{}{ "status": "ingested", @@ -2180,23 +2230,25 @@ func (s *Server) handleUpdateTaskList(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "Ungültiges JSON") return } + audienceBefore := s.store.GetTaskListUserIDs(listID) tl, err := s.store.UpdateTaskList(userID, listID, req.Title, req.Icon, req.Color, req.TeamID, req.SpaceID, req.Columns) if err != nil { writeStoreError(w, http.StatusBadRequest, err) return } - s.wsHub.BroadcastToAll("TASK_LIST_UPDATED", tl) + s.broadcastTaskListChange(listID, audienceBefore, "TASK_LIST_UPDATED") writeJSON(w, http.StatusOK, tl) } func (s *Server) handleDeleteTaskList(w http.ResponseWriter, r *http.Request) { userID := auth.GetUserID(r.Context()) listID := chi.URLParam(r, "id") + audienceBefore := s.store.GetTaskListUserIDs(listID) if err := s.store.DeleteTaskList(userID, listID); err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } - s.wsHub.BroadcastToAll("TASK_LIST_DELETED", map[string]string{"id": listID}) + s.broadcastTaskListChange(listID, audienceBefore, "TASK_LIST_DELETED") writeJSON(w, http.StatusOK, map[string]bool{"deleted": true}) } @@ -2245,8 +2297,7 @@ func (s *Server) handleCreateTaskListShare(w http.ResponseWriter, r *http.Reques writeError(w, http.StatusBadRequest, err.Error()) return } - updatedList, _ := s.store.GetTaskListByID(userID, listID) - s.wsHub.BroadcastToAll("TASK_LIST_UPDATED", updatedList) + s.broadcastTaskListChange(listID, nil, "TASK_LIST_UPDATED") writeJSON(w, http.StatusCreated, share) } @@ -2263,15 +2314,22 @@ func (s *Server) handleDeleteTaskListShare(w http.ResponseWriter, r *http.Reques writeError(w, http.StatusForbidden, "Nur der Eigentümer kann Freigaben entfernen") return } + audienceBefore := s.store.GetTaskListUserIDs(listID) if err := s.store.RemoveTaskListShare(listID, shareID); err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return } - updatedList, _ := s.store.GetTaskListByID(userID, listID) - s.wsHub.BroadcastToAll("TASK_LIST_UPDATED", updatedList) + s.broadcastTaskListChange(listID, audienceBefore, "TASK_LIST_UPDATED") writeJSON(w, http.StatusOK, map[string]bool{"deleted": true}) } +// broadcastTaskListChange notifies everyone who could see the list before or can see it now. Only the ID is sent, +// because the list payload depends on the receiver's permission; clients reload their lists. +func (s *Server) broadcastTaskListChange(listID string, audienceBefore []string, event string) { + users := append(audienceBefore, s.store.GetTaskListUserIDs(listID)...) + s.wsHub.BroadcastToUsers(users, event, map[string]string{"id": listID}) +} + // --- ORGANIZATIONS & TEAMS --- func (s *Server) handleGetOrganizations(w http.ResponseWriter, r *http.Request) { diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index 3ac0158..12cf1c6 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -4747,6 +4747,55 @@ func (s *Store) GetTaskListShares(taskListID string) ([]domain.TaskListShare, er return shares, nil } +// queryUserIDs runs a query returning a single user_id column and collects the distinct IDs. +func (s *Store) queryUserIDs(query string, args ...interface{}) []string { + rows, err := s.db.Query(query, args...) + if err != nil { + return []string{} + } + defer rows.Close() + ids := make([]string, 0) + seen := make(map[string]bool) + for rows.Next() { + var id string + if rows.Scan(&id) == nil && id != "" && !seen[id] { + seen[id] = true + ids = append(ids, id) + } + } + return ids +} + +// GetTaskListUserIDs returns every user who can see the task list: owner, direct shares, members of shared teams +// and members of the legacy team_id. +func (s *Store) GetTaskListUserIDs(listID string) []string { + if listID == "" { + return []string{} + } + return s.queryUserIDs( + `SELECT user_id FROM task_lists WHERE id = ? + UNION SELECT target_id FROM task_list_shares WHERE task_list_id = ? AND target_type = 'user' + UNION SELECT tm.user_id FROM team_members tm + WHERE tm.team_id IN (SELECT target_id FROM task_list_shares WHERE task_list_id = ? AND target_type = 'team') + OR tm.team_id IN (SELECT team_id FROM task_lists WHERE id = ? AND team_id != '')`, + listID, listID, listID, listID, + ) +} + +// GetSharedFolderUserIDs returns the owner of a shared workspace folder and everyone it is shared with. +func (s *Store) GetSharedFolderUserIDs(folderID string) []string { + if folderID == "" { + return []string{} + } + return s.queryUserIDs( + `SELECT owner_id FROM workspace_shared_folders WHERE id = ? + UNION SELECT target_id FROM workspace_folder_shares WHERE folder_id = ? AND target_type = 'user' + UNION SELECT tm.user_id FROM team_members tm + WHERE tm.team_id IN (SELECT target_id FROM workspace_folder_shares WHERE folder_id = ? AND target_type = 'team')`, + folderID, folderID, folderID, + ) +} + func (s *Store) getMembersForTaskList(tl *domain.TaskList) []domain.TaskListMember { memberMap := make(map[string]domain.TaskListMember) diff --git a/internal/ws/hub.go b/internal/ws/hub.go index 533815d..5f7ac91 100644 --- a/internal/ws/hub.go +++ b/internal/ws/hub.go @@ -93,6 +93,27 @@ type Hub struct { sseUserRooms map[string]map[*SSEClient]bool sseRooms map[string]map[*SSEClient]bool closed bool + + // authorizeRoom decides whether a user may receive events of a room; without it every join is denied + authorizeRoom func(userID, roomID string) bool +} + +// SetRoomAuthorizer installs the permission check used for every room join (URL room, JOIN_ROOM, SSE rooms). +func (h *Hub) SetRoomAuthorizer(f func(userID, roomID string) bool) { + h.mu.Lock() + h.authorizeRoom = f + h.mu.Unlock() +} + +// canJoin must be called without holding h.mu, since the authorizer queries the database. +func (h *Hub) canJoin(userID, roomID string) bool { + if roomID == "" || userID == "" || userID == "anonymous" { + return false + } + h.mu.RLock() + f := h.authorizeRoom + h.mu.RUnlock() + return f != nil && f(userID, roomID) } func NewHub() *Hub { @@ -107,6 +128,9 @@ func NewHub() *Hub { } func (h *Hub) Register(c *Client) { + if c.roomID != "" && !h.canJoin(c.userID, c.roomID) { + c.roomID = "" + } h.mu.Lock() h.allClients[c] = true if c.roomID != "" { @@ -167,6 +191,14 @@ func (h *Hub) Unregister(c *Client) { } func (h *Hub) RegisterSSE(c *SSEClient) { + if c.roomID != "" && !h.canJoin(c.userID, c.roomID) { + c.roomID = "" + } + for rID := range c.subscribedRooms { + if !h.canJoin(c.userID, rID) { + delete(c.subscribedRooms, rID) + } + } h.mu.Lock() h.sseClients[c] = true if c.roomID != "" { @@ -232,11 +264,15 @@ func (h *Hub) UnregisterSSE(c *SSEClient) { h.mu.Unlock() } -func (h *Hub) JoinSSERoomByID(sseID, roomID string) bool { +// JoinSSERoomByID adds the SSE connection sseID to a room; it only succeeds for the user owning that connection. +func (h *Hub) JoinSSERoomByID(sseID, userID, roomID string) bool { + if !h.canJoin(userID, roomID) { + return false + } h.mu.Lock() defer h.mu.Unlock() for client := range h.sseClients { - if client.id == sseID { + if client.id == sseID && client.userID == userID { if h.sseRooms[roomID] == nil { h.sseRooms[roomID] = make(map[*SSEClient]bool) } @@ -251,11 +287,11 @@ func (h *Hub) JoinSSERoomByID(sseID, roomID string) bool { return false } -func (h *Hub) LeaveSSERoomByID(sseID, roomID string) bool { +func (h *Hub) LeaveSSERoomByID(sseID, userID, roomID string) bool { h.mu.Lock() defer h.mu.Unlock() for client := range h.sseClients { - if client.id == sseID { + if client.id == sseID && client.userID == userID { if room, ok := h.sseRooms[roomID]; ok { delete(room, client) if len(room) == 0 { @@ -305,32 +341,6 @@ func (h *Hub) Close() { // Broadcasting Methods -func (h *Hub) BroadcastToAll(event string, payload interface{}) { - msg := domain.WSMessage{ - Event: event, - Payload: payload, - Timestamp: domain.NowISO(), - } - data, err := json.Marshal(msg) - if err != nil { - return - } - - h.mu.RLock() - if h.closed { - h.mu.RUnlock() - return - } - defer h.mu.RUnlock() - - for client := range h.allClients { - client.trySend(data) - } - for sseClient := range h.sseClients { - sseClient.trySend(data) - } -} - func (h *Hub) BroadcastToUser(userID, event string, payload interface{}) { msg := domain.WSMessage{ Event: event, @@ -361,15 +371,21 @@ func (h *Hub) BroadcastToUser(userID, event string, payload interface{}) { } } -func (h *Hub) BroadcastTaskEvent(userID, event string, payload interface{}) { - h.BroadcastToUser(userID, event, payload) +// BroadcastToUsers sends one event to every connection of the given users (duplicates are ignored). +func (h *Hub) BroadcastToUsers(userIDs []string, event string, payload interface{}) { + seen := make(map[string]bool, len(userIDs)) + for _, uid := range userIDs { + if uid == "" || seen[uid] { + continue + } + seen[uid] = true + h.BroadcastToUser(uid, event, payload) + } } func (h *Hub) BroadcastDocEvent(docID, event string, payload interface{}) { if docID != "" { h.BroadcastRoom("doc:"+docID, event, payload, nil) - } else { - h.BroadcastToAll(event, payload) } } @@ -413,7 +429,10 @@ func (h *Hub) BroadcastListEvent(listID, event string, payload interface{}) { h.Broadcast("", listID, event, payload) } -func (h *Hub) JoinRoom(c *Client, roomID string) { +func (h *Hub) JoinRoom(c *Client, roomID string) bool { + if !h.canJoin(c.userID, roomID) { + return false + } h.mu.Lock() defer h.mu.Unlock() if h.rooms[roomID] == nil { @@ -424,6 +443,7 @@ func (h *Hub) JoinRoom(c *Client, roomID string) { c.subscribedRooms = make(map[string]bool) } c.subscribedRooms[roomID] = true + return true } func (h *Hub) LeaveRoom(c *Client, roomID string) { @@ -512,11 +532,13 @@ func (h *Hub) ServeSSE(w http.ResponseWriter, r *http.Request, roomID, userID st return } + // The server's WriteTimeout would cut long-lived streams; lift it for this response only + _ = http.NewResponseController(w).SetWriteDeadline(time.Time{}) + w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache, no-transform") w.Header().Set("Connection", "keep-alive") w.Header().Set("X-Accel-Buffering", "no") - w.Header().Set("Access-Control-Allow-Origin", "*") client := &SSEClient{ id: domain.NewID("sse"), @@ -591,7 +613,7 @@ func (c *Client) readPump() { break } - // Handle client-initiated lock, CRDT, room subscription, and presence messages + // Handle client-initiated room subscription messages var incoming struct { Event string `json:"event"` DocID string `json:"docId"` diff --git a/internal/ws/hub_test.go b/internal/ws/hub_test.go index eb47855..e107485 100644 --- a/internal/ws/hub_test.go +++ b/internal/ws/hub_test.go @@ -22,7 +22,7 @@ func TestHubCloseAndGracefulShutdown(t *testing.T) { } // Verify broadcast works - hub.BroadcastToAll("TEST_EVENT", map[string]string{"msg": "hello"}) + hub.BroadcastToUser("user-1", "TEST_EVENT", map[string]string{"msg": "hello"}) // Concurrently invoke Close() and Broadcast() to verify no panics or deadlocks var wg sync.WaitGroup @@ -31,7 +31,7 @@ func TestHubCloseAndGracefulShutdown(t *testing.T) { go func() { defer wg.Done() for i := 0; i < 50; i++ { - hub.BroadcastToAll("SPAM", map[string]int{"i": i}) + hub.BroadcastToUser("user-1", "SPAM", map[string]int{"i": i}) time.Sleep(1 * time.Millisecond) } }() @@ -56,3 +56,37 @@ func TestHubCloseAndGracefulShutdown(t *testing.T) { } } } + +func TestRoomJoinsRequireAuthorization(t *testing.T) { + hub := NewHub() + c := &Client{hub: hub, userID: "user-1", send: make(chan []byte, 10)} + + if hub.JoinRoom(c, "lst_1") { + t.Fatal("join must be denied without an authorizer") + } + + hub.SetRoomAuthorizer(func(userID, roomID string) bool { + return userID == "user-1" && roomID == "lst_allowed" + }) + if hub.JoinRoom(c, "lst_other") { + t.Fatal("join to an unauthorized room must be denied") + } + if !hub.JoinRoom(c, "lst_allowed") { + t.Fatal("join to an authorized room must succeed") + } + + anon := &Client{hub: hub, userID: "anonymous", send: make(chan []byte, 10)} + if hub.JoinRoom(anon, "lst_allowed") { + t.Fatal("anonymous clients must never join rooms") + } + + sse := &SSEClient{id: "sse_1", userID: "user-1", send: make(chan []byte, 10), + subscribedRooms: map[string]bool{"lst_allowed": true, "doc:secret": true}} + hub.RegisterSSE(sse) + if sse.subscribedRooms["doc:secret"] { + t.Fatal("unauthorized initial SSE rooms must be dropped") + } + if hub.JoinSSERoomByID("sse_1", "user-2", "lst_allowed") { + t.Fatal("joining with another user's sseId must fail") + } +}