fix(trash): owner-scope all trash ops — close cross-tenant IDOR/disclosure

Drift-audit Group 1 (authz/IDOR). Multi-user is live, so these were
exploitable ACL bypasses:

- trash.py: add _owner_clause() and apply it to _exists_alive, restore,
  purge, list_trash, and purge_expired. A batch_id is a bearer token;
  without an owner predicate a leaked/guessed id let one tenant read
  (list_trash), restore, or PERMANENTLY purge another's content. Topics
  and rules carried no owner check at all (_OWNER mapped them to None) —
  ownership now derives through the parent rulebook (or owning project,
  for project-scoped rules).
- purge_expired is now per-user; trash_scheduler iterates every user and
  applies that user's own trash_retention_days window, instead of
  applying user 1's window to everyone (early data loss for other users).
- rulebooks subscribe/unsubscribe_project now assert project ownership,
  matching the suppression endpoints.
- topic/rule DELETE routes return 404 when nothing owned was removed.

Regression test locks in that every model — including topics/rules —
gets a real owner clause.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/fabledassistant/services/trash.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 import uuid
 from datetime import datetime, timezone
 
-from sqlalchemy import select, update
+from sqlalchemy import or_, select, update
 
 from fabledassistant.models import async_session
 from fabledassistant.models.note import Note
@@ -19,19 +19,50 @@ from fabledassistant.models.project import Project
 from fabledassistant.models.milestone import Milestone
 from fabledassistant.models.rulebook import Rulebook, RulebookTopic, Rule
 
-# entity_type -> (Model, owner_column_name or None). Used by the existence check.
-_OWNER = {
-    "note": (Note, "user_id"),
-    "task": (Note, "user_id"),
-    "event": (Event, "user_id"),
-    "project": (Project, "user_id"),
-    "milestone": (Milestone, "user_id"),
-    "rulebook": (Rulebook, "owner_user_id"),
-    "topic": (RulebookTopic, None),
-    "rule": (Rule, None),
+# entity_type -> Model. Used to resolve which table a trash op targets.
+_MODEL_FOR = {
+    "note": Note,
+    "task": Note,
+    "event": Event,
+    "project": Project,
+    "milestone": Milestone,
+    "rulebook": Rulebook,
+    "topic": RulebookTopic,
+    "rule": Rule,
 }
 
 
+def _owner_clause(model, user_id: int):
+    """Boolean expr scoping `model` rows to the ones `user_id` owns.
+
+    EVERY trash query (exists-check, restore, purge, list, retention sweep)
+    must carry this — a batch_id is a bearer token, so without an owner
+    predicate a leaked/guessed id lets one tenant read, restore, or
+    permanently destroy another's content. Topics and rules carry no
+    user_id of their own; ownership is derived through the parent rulebook
+    (or, for project-scoped rules, the owning project).
+    """
+    if model is Rulebook:
+        return Rulebook.owner_user_id == user_id
+    if model is RulebookTopic:
+        return RulebookTopic.rulebook_id.in_(
+            select(Rulebook.id).where(Rulebook.owner_user_id == user_id)
+        )
+    if model is Rule:
+        return or_(
+            Rule.topic_id.in_(
+                select(RulebookTopic.id)
+                .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+                .where(Rulebook.owner_user_id == user_id)
+            ),
+            Rule.project_id.in_(
+                select(Project.id).where(Project.user_id == user_id)
+            ),
+        )
+    # Note, Event, Project, Milestone all carry user_id directly.
+    return model.user_id == user_id
+
+
 async def _set(session, model, where, batch, now) -> None:
     """Stamp deleted_at + batch on live rows matching `where`."""
     await session.execute(
@@ -42,10 +73,8 @@ async def _set(session, model, where, batch, now) -> None:
 
 
 async def _exists_alive(session, user_id: int, etype: str, eid: int) -> bool:
-    model, owner = _OWNER[etype]
-    where = [model.id == eid, model.deleted_at.is_(None)]
-    if owner:
-        where.append(getattr(model, owner) == user_id)
+    model = _MODEL_FOR[etype]
+    where = [model.id == eid, model.deleted_at.is_(None), _owner_clause(model, user_id)]
     return (await session.execute(select(model.id).where(*where))).first() is not None
 
 
@@ -135,7 +164,7 @@ async def restore(user_id: int, batch_id: str) -> int:
         for model in _ALL:
             res = await session.execute(
                 update(model)
-                .where(model.deleted_batch_id == batch_id)
+                .where(model.deleted_batch_id == batch_id, _owner_clause(model, user_id))
                 .values(deleted_at=None, deleted_batch_id=None)
             )
             n += res.rowcount or 0
@@ -150,7 +179,9 @@ async def purge(user_id: int, batch_id: str) -> int:
     async with async_session() as session:
         for model in _ALL:
             res = await session.execute(
-                sql_delete(model).where(model.deleted_batch_id == batch_id)
+                sql_delete(model).where(
+                    model.deleted_batch_id == batch_id, _owner_clause(model, user_id)
+                )
             )
             n += res.rowcount or 0
         await session.commit()
@@ -163,7 +194,9 @@ async def list_trash(user_id: int) -> list[dict]:
     async with async_session() as session:
         for model in _ALL:
             rows = (await session.execute(
-                select(model).where(model.deleted_at.isnot(None))
+                select(model).where(
+                    model.deleted_at.isnot(None), _owner_clause(model, user_id)
+                )
             )).scalars().all()
             for r in rows:
                 grp = batches.setdefault(
@@ -185,9 +218,12 @@ async def list_trash(user_id: int) -> list[dict]:
     return out
 
 
-async def purge_expired(retention_days: int) -> int:
-    """Cron entry: hard-delete rows trashed more than retention_days ago.
+async def purge_expired(user_id: int, retention_days: int) -> int:
+    """Cron entry: hard-delete THIS user's rows trashed more than retention_days ago.
 
+    Scoped to one owner so the scheduler can apply each user's own
+    `trash_retention_days` window — a single global sweep would let one
+    user's short window prematurely destroy another's data.
     retention_days <= 0 disables auto-purge (returns 0 without touching anything).
     """
     from datetime import timedelta
@@ -200,7 +236,9 @@ async def purge_expired(retention_days: int) -> int:
         for model in _ALL:
             res = await session.execute(
                 sql_delete(model).where(
-                    model.deleted_at.isnot(None), model.deleted_at < cutoff
+                    model.deleted_at.isnot(None),
+                    model.deleted_at < cutoff,
+                    _owner_clause(model, user_id),
                 )
             )
             n += res.rowcount or 0