fix(realtime): scope events to authorized users, authorize rooms, keep SSE alive

Phase 0 of the collaboration plan: - Remove BroadcastToAll; task, task list and workspace folder events now only reach users who can see the resource. Task list events carry only the list ID, clients reload with their own permissions. - Reject unauthenticated WebSocket/SSE connections (401). Every room join (URL room, JOIN_ROOM, SSE rooms) is checked via a room authorizer; /events/rooms requires login and only accepts the caller's own sseId. - Reorder rejects tasks the caller cannot access or edit (IDOR). - Lift the server WriteTimeout for SSE streams so they are not cut after 30s. - Fix folder share removal sending DOC_DELETED to the share ID. - Tests for all of the above; docs and flaw list updated. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-28 11:19:14 +01:00
1 parent 74797e3f44
commit bcc06da3ee
@@ -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)
+	}
+}