SECURITY: notifications read paths ignore user_id — list/list_archived/unread_count return every user's rows and per-id read/dismiss/archive act cross-user (CWE-862, found on #2716 review)

changelog.d/tsk-7uxooi-notifications-user-scope.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Cross-user notification read and mutation: `list`, `list_archived`, `unread_count`, `mark_read`, `archive`, and `mark_all_read` now scope to the authenticated user (`user_id IS NULL OR user_id = ?`), so a user can only see and modify their own notifications plus broadcasts. Previously these endpoints returned every user's rows and allowed cross-user mutations (CWE-862).

tests/test_notifications_user_scope.py

@@ -0,0 +1,219 @@
+import secrets
+
+import pytest
+import pytest_asyncio
+import yaml
+from httpx import ASGITransport, AsyncClient
+
+from tinyagentos.app import create_app
+from tinyagentos.notifications import NotificationStore
+from taos_test_csrf import csrf_event_hooks
+
+
+def _make_config(tmp_path) -> dict:
+    return {
+        "server": {"host": "0.0.0.0", "port": 6969},
+        "backends": [],
+        "qmd": {"url": "http://localhost:7832"},
+        "agents": [],
+        "metrics": {"poll_interval": 30, "retention_days": 30},
+    }
+
+
+@pytest_asyncio.fixture
+async def notif_store(tmp_path):
+    store = NotificationStore(tmp_path / "notifications.db")
+    await store.init()
+    yield store
+    await store.close()
+
+
+@pytest.mark.asyncio
+class TestNotificationStoreUserScope:
+    async def test_list_returns_own_and_broadcast(self, notif_store):
+        await notif_store.add("a", "a msg", user_id="u1")
+        await notif_store.add("b", "b msg", user_id="u2")
+        await notif_store.add("c", "c msg")
+        items = await notif_store.list(user_id="u1")
+        titles = {i["title"] for i in items}
+        assert titles == {"a", "c"}
+
+    async def test_list_excludes_other_users(self, notif_store):
+        await notif_store.add("a", "a msg", user_id="u1")
+        await notif_store.add("b", "b msg", user_id="u2")
+        items = await notif_store.list(user_id="u1")
+        assert all(i["user_id"] != "u2" for i in items)
+
+    async def test_list_archived_returns_own_only(self, notif_store):
+        await notif_store.add("a", "a msg", user_id="u1")
+        await notif_store.add("b", "b msg", user_id="u2")
+        await notif_store.add("c", "c msg")
+        items = await notif_store.list()
+        await notif_store.archive(items[0]["id"], user_id="u1")
+        await notif_store.archive(items[1]["id"], user_id="u2")
+        history = await notif_store.list_archived(user_id="u1")
+        titles = {h["title"] for h in history}
+        assert titles == {"a"}
+
+    async def test_unread_count_counts_own_and_broadcast(self, notif_store):
+        await notif_store.add("a", "a msg", user_id="u1")
+        await notif_store.add("b", "b msg", user_id="u2")
+        await notif_store.add("c", "c msg")
+        assert await notif_store.unread_count(user_id="u1") == 2
+
+    async def test_none_user_id_returns_unfiltered(self, notif_store):
+        await notif_store.add("a", "a msg", user_id="u1")
+        await notif_store.add("b", "b msg", user_id="u2")
+        items = await notif_store.list(user_id=None)
+        assert len(items) == 2
+
+    async def test_mark_read_scoped_to_user(self, notif_store):
+        await notif_store.add("a", "a msg", user_id="u1")
+        await notif_store.add("b", "b msg", user_id="u2")
+        u1_items = await notif_store.list(user_id="u1")
+        u2_items = await notif_store.list(user_id="u2")
+        u1_id = u1_items[0]["id"]
+        u2_id = u2_items[0]["id"]
+        affected = await notif_store.mark_read(u2_id, user_id="u1")
+        assert affected == 0
+        assert (await notif_store.list(user_id="u2"))[0]["read"] is False
+        affected = await notif_store.mark_read(u1_id, user_id="u1")
+        assert affected == 1
+        assert (await notif_store.list(user_id="u1"))[0]["read"] is True
+
+    async def test_archive_scoped_to_user(self, notif_store):
+        await notif_store.add("a", "a msg", user_id="u1")
+        await notif_store.add("b", "b msg", user_id="u2")
+        u1_items = await notif_store.list(user_id="u1")
+        u2_items = await notif_store.list(user_id="u2")
+        u1_id = u1_items[0]["id"]
+        u2_id = u2_items[0]["id"]
+        affected = await notif_store.archive(u2_id, user_id="u1")
+        assert affected == 0
+        assert len(await notif_store.list_archived(user_id="u2")) == 0
+        affected = await notif_store.archive(u1_id, user_id="u1")
+        assert affected == 1
+        assert len(await notif_store.list_archived(user_id="u1")) == 1
+
+
+@pytest.mark.asyncio
+class TestNotificationRoutesUserScope:
+    @pytest_asyncio.fixture
+    async def two_user_app(self, tmp_path):
+        config = _make_config(tmp_path)
+        (tmp_path / "config.yaml").write_text(yaml.dump(config))
+        (tmp_path / ".setup_complete").touch()
+
+        app = create_app(data_dir=tmp_path)
+
+        notif_store = app.state.notifications
+        if notif_store._db is not None:
+            await notif_store.close()
+        await notif_store.init()
+
+        auth = app.state.auth
+        auth.setup_user("alice", "Alice", "", "alicepass123")
+        alice_rec = auth.find_user("alice")
+        alice_token = auth.create_session(user_id=alice_rec["id"], long_lived=True)
+
+        bob_invite = auth.add_user_invite("bob", "alice")
+        auth.complete_invite("bob", bob_invite, "Bob", "", "bobpass123")
+        bob_rec = auth.find_user("bob")
+        bob_token = auth.create_session(user_id=bob_rec["id"], long_lived=True)
+
+        app.state._startup_complete = True
+
+        return app, alice_rec["id"], alice_token, bob_rec["id"], bob_token
+
+    async def _alice_client(self, app, alice_token):
+        transport = ASGITransport(app=app)
+        return AsyncClient(
+            transport=transport,
+            base_url="http://test",
+            cookies={"taos_session": alice_token},
+            event_hooks=csrf_event_hooks(),
+        )
+
+    async def _bob_client(self, app, bob_token):
+        transport = ASGITransport(app=app)
+        return AsyncClient(
+            transport=transport,
+            base_url="http://test",
+            cookies={"taos_session": bob_token},
+            event_hooks=csrf_event_hooks(),
+        )
+
+    async def test_list_excludes_other_user(self, two_user_app):
+        app, alice_id, alice_token, bob_id, bob_token = two_user_app
+        store = app.state.notifications
+        await store.add("alice-notif", "for alice", user_id=alice_id)
+        await store.add("bob-notif", "for bob", user_id=bob_id)
+        await store.add("broadcast", "for everyone")
+        async with await self._alice_client(app, alice_token) as c:
+            resp = await c.get("/api/notifications")
+            assert resp.status_code == 200
+            data = resp.json()
+            titles = {i["title"] for i in data}
+            assert "alice-notif" in titles
+            assert "bob-notif" not in titles
+            assert "broadcast" in titles
+
+    async def test_archived_excludes_other_user(self, two_user_app):
+        app, alice_id, alice_token, bob_id, bob_token = two_user_app
+        store = app.state.notifications
+        await store.add("alice-notif", "for alice", user_id=alice_id)
+        await store.add("bob-notif", "for bob", user_id=bob_id)
+        alice_items = await store.list(user_id=alice_id)
+        bob_items = await store.list(user_id=bob_id)
+        await store.archive(alice_items[0]["id"], user_id=alice_id)
+        await store.archive(bob_items[0]["id"], user_id=bob_id)
+        async with await self._alice_client(app, alice_token) as c:
+            resp = await c.get("/api/notifications/archived")
+            assert resp.status_code == 200
+            data = resp.json()
+            titles = {i["title"] for i in data}
+            assert "alice-notif" in titles
+            assert "bob-notif" not in titles
+
+    async def test_count_excludes_other_user(self, two_user_app):
+        app, alice_id, alice_token, bob_id, bob_token = two_user_app
+        store = app.state.notifications
+        await store.add("alice-notif", "for alice", user_id=alice_id)
+        await store.add("bob-notif", "for bob", user_id=bob_id)
+        async with await self._alice_client(app, alice_token) as c:
+            resp = await c.get("/api/notifications/count")
+            assert resp.status_code == 200
+            assert "1" in resp.text
+
+    async def test_mark_read_other_user_returns_404(self, two_user_app):
+        app, alice_id, alice_token, bob_id, bob_token = two_user_app
+        store = app.state.notifications
+        await store.add("bob-notif", "for bob", user_id=bob_id)
+        bob_items = await store.list(user_id=bob_id)
+        bob_notif_id = bob_items[0]["id"]
+        async with await self._alice_client(app, alice_token) as c:
+            resp = await c.post(f"/api/notifications/{bob_notif_id}/read")
+            assert resp.status_code == 404
+            assert (await store.list(user_id=bob_id))[0]["read"] is False
+
+    async def test_archive_other_user_returns_404(self, two_user_app):
+        app, alice_id, alice_token, bob_id, bob_token = two_user_app
+        store = app.state.notifications
+        await store.add("bob-notif", "for bob", user_id=bob_id)
+        bob_items = await store.list(user_id=bob_id)
+        bob_notif_id = bob_items[0]["id"]
+        async with await self._alice_client(app, alice_token) as c:
+            resp = await c.post(f"/api/notifications/{bob_notif_id}/archive")
+            assert resp.status_code == 404
+            assert len(await store.list_archived(user_id=bob_id)) == 0
+
+    async def test_mark_own_notification_succeeds(self, two_user_app):
+        app, alice_id, alice_token, bob_id, bob_token = two_user_app
+        store = app.state.notifications
+        await store.add("alice-notif", "for alice", user_id=alice_id)
+        alice_items = await store.list(user_id=alice_id)
+        alice_notif_id = alice_items[0]["id"]
+        async with await self._alice_client(app, alice_token) as c:
+            resp = await c.post(f"/api/notifications/{alice_notif_id}/read")
+            assert resp.status_code == 200
+            assert (await store.list(user_id=alice_id))[0]["read"] is True

tinyagentos/notifications.py

@@ -211,47 +211,86 @@ async def add(
             except Exception:
                 logger.warning("NotificationStore: could not schedule web-push", exc_info=True)
 
-    async def list(self, limit: int = 20, unread_only: bool = False) -> list[dict]:
+    async def list(
+        self,
+        limit: int = 20,
+        unread_only: bool = False,
+        user_id: str | None = None,
+    ) -> list[dict]:
         # Active feed: archived (dismissed) notifications are excluded.
         conds = ["archived = 0"]
+        if user_id is not None:
+            conds.append("(user_id IS NULL OR user_id = ?)")
         if unread_only:
             conds.append("read = 0")
+        params: tuple = (user_id, limit) if user_id is not None else (limit,)
         sql = (
             "SELECT id, timestamp, level, title, message, read, source, data, user_id FROM notifications"
             f" WHERE {' AND '.join(conds)} ORDER BY timestamp DESC LIMIT ?"
         )
-        async with self._db.execute(sql, (limit,)) as cursor:
+        async with self._db.execute(sql, params) as cursor:
             rows = await cursor.fetchall()
         return [_serialize_row(r) for r in rows]
 
-    async def list_archived(self, limit: int = 50) -> list[dict]:
+    async def list_archived(
+        self,
+        limit: int = 50,
+        user_id: str | None = None,
+    ) -> list[dict]:
         # History view: the dismissed notifications, newest first. Nothing is
         # deleted, so this is the durable record (#62 / append-only #103).
+        conds = ["archived = 1"]
+        if user_id is not None:
+            conds.append("(user_id IS NULL OR user_id = ?)")
+        params: tuple = (user_id, limit) if user_id is not None else (limit,)
         async with self._db.execute(
             "SELECT id, timestamp, level, title, message, read, source, data, user_id FROM notifications"
-            " WHERE archived = 1 ORDER BY timestamp DESC LIMIT ?",
-            (limit,),
+            f" WHERE {' AND '.join(conds)} ORDER BY timestamp DESC LIMIT ?",
+            params,
         ) as cursor:
             rows = await cursor.fetchall()
         return [_serialize_row(r) for r in rows]
 
-    async def unread_count(self) -> int:
+    async def unread_count(self, user_id: str | None = None) -> int:
+        conds = ["read = 0", "archived = 0"]
+        if user_id is not None:
+            conds.append("(user_id IS NULL OR user_id = ?)")
+        params: tuple = (user_id,) if user_id is not None else ()
         async with self._db.execute(
-            "SELECT COUNT(*) FROM notifications WHERE read = 0 AND archived = 0"
+            f"SELECT COUNT(*) FROM notifications WHERE {' AND '.join(conds)}",
+            params,
         ) as cursor:
             row = await cursor.fetchone()
         return row[0] if row else 0
 
-    async def mark_read(self, notif_id: int) -> None:
-        await self._db.execute("UPDATE notifications SET read = 1 WHERE id = ?", (notif_id,))
+    async def mark_read(self, notif_id: int, user_id: str | None = None) -> int:
+        if user_id is not None:
+            cursor = await self._db.execute(
+                "UPDATE notifications SET read = 1 WHERE id = ? AND (user_id IS NULL OR user_id = ?)",
+                (notif_id, user_id),
+            )
+        else:
+            # Internal/system caller: unfiltered update.
+            cursor = await self._db.execute(
+                "UPDATE notifications SET read = 1 WHERE id = ?", (notif_id,)
+            )
         await self._db.commit()
+        return cursor.rowcount
 
-    async def archive(self, notif_id: int) -> None:
+    async def archive(self, notif_id: int, user_id: str | None = None) -> int:
         # Dismiss = archive. The row stays; the History view still shows it.
-        await self._db.execute(
-            "UPDATE notifications SET archived = 1 WHERE id = ?", (notif_id,)
-        )
+        if user_id is not None:
+            cursor = await self._db.execute(
+                "UPDATE notifications SET archived = 1 WHERE id = ? AND (user_id IS NULL OR user_id = ?)",
+                (notif_id, user_id),
+            )
+        else:
+            # Internal/system caller: unfiltered update.
+            cursor = await self._db.execute(
+                "UPDATE notifications SET archived = 1 WHERE id = ?", (notif_id,)
+            )
         await self._db.commit()
+        return cursor.rowcount
 
     async def archive_by_source_ref(self, source: str, request_id) -> int:
         """Archive active notifications whose JSON `data.request_id` matches.
@@ -262,6 +301,7 @@ async def archive_by_source_ref(self, source: str, request_id) -> int:
         (#62: nothing is deleted). Idempotent: rows already archived are
         skipped; returns the number newly archived.
         """
+        # Intentionally global: resolves by source + request_id, not by user.
         async with self._db.execute(
             "SELECT id, data FROM notifications WHERE source = ? AND archived = 0",
             (source,),
@@ -279,6 +319,8 @@ async def archive_by_source_ref(self, source: str, request_id) -> int:
             if str(payload.get("request_id")) == target:
                 ids.append(nid)
         if ids:
+            # Intentionally global: source-ref resolution applies to the row,
+            # not to a specific user.
             placeholders = ",".join("?" * len(ids))
             await self._db.execute(
                 f"UPDATE notifications SET archived = 1, read = 1 WHERE id IN ({placeholders})",
@@ -287,15 +329,23 @@ async def archive_by_source_ref(self, source: str, request_id) -> int:
             await self._db.commit()
         return len(ids)
 
-    async def mark_all_read(self) -> int:
-        cursor = await self._db.execute("UPDATE notifications SET read = 1 WHERE read = 0")
+    async def mark_all_read(self, user_id: str | None = None) -> int:
+        if user_id is not None:
+            cursor = await self._db.execute(
+                "UPDATE notifications SET read = 1 WHERE read = 0 AND (user_id IS NULL OR user_id = ?)",
+                (user_id,),
+            )
+        else:
+            # Internal/system caller: unfiltered update.
+            cursor = await self._db.execute("UPDATE notifications SET read = 1 WHERE read = 0")
         await self._db.commit()
         return cursor.rowcount
 
     async def cleanup(self, max_age_days: int = 30) -> int:
         # Age out only old UNdismissed notifications. Archived rows are the
         # durable history a user explicitly dismissed (#62 / append-only #103),
         # so they are never GC'd here.
+        # Intentionally global: retention/prune is a system-wide operation.
         cutoff = int(time.time()) - (max_age_days * 86400)
         cursor = await self._db.execute(
             "DELETE FROM notifications WHERE timestamp < ? AND archived = 0", (cutoff,)