sprint2+3: remaining services visibility filter + search provider permission-aware + dashboard route

RBAC_PROGRESS.md

@@ -1,36 +1,29 @@
 # RBAC Build Progress — LeoCRM
 
-## Letztes Update: 2026-07-29 01:31 CEST
+## Letztes Update: 2026-07-29 02:00 CEST
 
 ## Sprint 1 — Fundament (14h) ✅ VOLLSTÄNDIG
 
-### Alle Items erledigt:
-- [x] EntityPermission Model (`app/models/entity_permission.py`)
-- [x] OwnedMixin (`app/models/owned_mixin.py`)
-- [x] Migration 0049: entity_permissions Tabelle — ✅ Produktion
-- [x] Migration 0050: owner_id auf 15 Tabellen — ✅ Produktion
-- [x] Migration 0051: Folder ACLs → entity_permissions — ✅ Produktion
-- [x] Migration 0052: RLS Policies auf contacts — ✅ Produktion
-- [x] OwnedMixin auf 13 Models angewendet
-- [x] Universeller Permission Service (648 Zeilen)
-- [x] Universelle Permission API (6 Endpoints + Rate Limiting)
-- [x] set_user_context() für RLS in db/__init__.py
-- [x] deps.py: set_user_context() wird bei jedem Request aufgerufen
-- [x] Rate Limiting auf Permission-Änderungen (50/min/User)
-- [x] Container neu gestartet, alles deployed
-- [x] Git committed und gepusht (ea1c1d5)
+Alle Items erledigt und in Produktion. Siehe vorherige Commits.
 
-## Sprint 2 — Row-Level Security (16h) 🔄 NÄCHSTER
+## Sprint 2 — Row-Level Security (16h) ✅ GRÖSSTEILS FERTIG
 
-### Geplante Items:
-- [ ] apply_visibility_filter() Helper
-- [ ] Query-Filter in alle 28 Routes
-- [ ] Child-Entity-Vererbung
-- [ ] Batch-Resolution in Listen-Queries
-- [ ] BaseSearchProvider mit Visibility-Filter
-- [ ] ContactDetail/ContactsList Permission-Checks (Frontend)
+### Erledigt ✅
+- [x] apply_visibility_filter() Helper (`app/core/visibility.py`, 237 Zeilen)
+- [x] Query-Filter in contact_service (list, get, create, update, delete, export)
+- [x] Query-Filter in 8 weiteren Services (address, attachment, bank_account, workflow, sequence, saved_filter, saved_view, webhook)
+- [x] 8 Routes angepasst (user_id + is_system_admin + PermissionError handling)
+- [x] BaseSearchProvider mit Visibility-Filter (`app/plugins/builtins/unified_search/base_provider.py`)
+- [x] Frontend Permission-Checks: ContactDetailPage + ContactDetail + ContactsList
+- [x] Field-Level UI: hidden fields nicht gerendert, readonly disabled
+- [x] OwnedMixin auf 8 Models korrigiert (Klassendefinition)
+- [x] Alles deployed und committed (52a5c34)
+
+### Noch offen ⬜
+- [ ] Child-Entity-Vererbung (ContactPerson erbt von Contact)
 - [ ] Copy/Duplicate Permission
-- [ ] EXISTS-Optimization für RLS
+- [ ] EXISTS-Optimization für RLS (bereits in visibility.py verwendet)
+- [ ] Weitere 12 Routes (notifications, entity_history, custom_fields, etc.)
 
 ## Sprint 3-23 ⬜
 
@@ -52,12 +45,17 @@
 | 5afa1fa | sprint1: entity_permissions + owned_mixin + service + API + migrations 0049+0050 |
 | 48647a5 | sprint1: set_user_context + RLS policies + folder ACL migration 0051+0052 |
 | ea1c1d5 | sprint1 complete: rate limiting on permission changes |
+| 479ee04 | sprint2: visibility filter + contact service access checks |
+| 9fc84b7 | sprint2: 8 services + 8 routes visibility filter + BaseSearchProvider |
+| 52a5c34 | sprint2: frontend permission checks for ContactDetail + ContactsList |
 
 ## Was in Produktion läuft
 - entity_permissions Tabelle (universelle ACLs)
-- owner_id auf 15 Tabellen (contacts, addresses, attachments, etc.)
-- PostgreSQL RLS auf contacts (4 Policies: admin, owner, tenant-owned, shared)
-- set_user_context() wird bei jedem Request gesetzt
+- owner_id auf 15 Tabellen
+- PostgreSQL RLS auf contacts (4 Policies)
+- set_user_context() bei jedem Request
 - Universelle Permission API unter /api/v1/permissions/*
 - Rate Limiting auf Permission-Änderungen
-- Folder ACLs in entity_permissions migriert
+- Visibility Filter in 9 Services (contacts + 8 weitere)
+- Frontend Permission-Checks in ContactDetail + ContactsList
+- BaseSearchProvider für Permission-aware Search

app/models/custom_field_definition.py

@@ -13,7 +13,7 @@ from app.core.db import Base, TenantMixin
 from app.models.owned_mixin import OwnedMixin
 
 
-class CustomFieldDefinition(Base, TenantMixin):
+class CustomFieldDefinition(Base, TenantMixin, OwnedMixin):
     """User-defined custom field definition stored in the database.
 
     These definitions are merged with plugin-provided custom fields

app/models/entity_history.py

@@ -15,7 +15,7 @@ from app.core.db import Base, TenantMixin
 from app.models.owned_mixin import OwnedMixin
 
 
-class EntityHistory(Base, TenantMixin):
+class EntityHistory(Base, TenantMixin, OwnedMixin):
     """Snapshot history for undo/restore functionality.
 
     Every CRUD action (create/update/delete) stores a full entity snapshot

app/plugins/builtins/unified_search/providers/contact_provider.py

@@ -9,71 +9,124 @@ from typing import Any
 from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider
+
 logger = logging.getLogger(__name__)
 
 
-class ContactSearchProvider:
+class ContactSearchProvider(BaseSearchProvider):
     """Search provider for Contact entities (all types)."""
 
     entity_type = "contact"
 
-    async def search_fts(
+    async def _search_fts_filtered(
         self,
         db: AsyncSession,
         tsquery: str,
         tenant_id: uuid.UUID,
         limit: int,
+        visible_ids: set[uuid.UUID] | None,
     ) -> list[dict[str, Any]]:
-        """Full-text search on contacts.search_tsv."""
-        sql = text(
-            """
-            SELECT c.*, ts_rank(c.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank
-            FROM contacts c
-            WHERE c.tenant_id = :tid
-              AND c.deleted_at IS NULL
-              AND c.search_tsv @@ to_tsquery('pg_catalog.german', :q)
-            ORDER BY rank DESC
-            LIMIT :lim
-            """
-        )
-        result = await db.execute(
-            sql,
-            {"q": tsquery, "tid": tenant_id, "lim": limit},
-        )
+        """Full-text search on contacts.search_tsv, filtered by visible_ids.
+
+        If visible_ids is None, no visibility filter is applied (system admin).
+        """
+        if visible_ids is not None:
+            sql = text(
+                """
+                SELECT c.*, ts_rank(c.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank
+                FROM contacts c
+                WHERE c.tenant_id = :tid
+                  AND c.deleted_at IS NULL
+                  AND c.search_tsv @@ to_tsquery('pg_catalog.german', :q)
+                  AND c.id = ANY(:visible_ids)
+                ORDER BY rank DESC
+                LIMIT :lim
+                """
+            )
+            result = await db.execute(
+                sql,
+                {
+                    "q": tsquery,
+                    "tid": tenant_id,
+                    "lim": limit,
+                    "visible_ids": list(visible_ids),
+                },
+            )
+        else:
+            sql = text(
+                """
+                SELECT c.*, ts_rank(c.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank
+                FROM contacts c
+                WHERE c.tenant_id = :tid
+                  AND c.deleted_at IS NULL
+                  AND c.search_tsv @@ to_tsquery('pg_catalog.german', :q)
+                ORDER BY rank DESC
+                LIMIT :lim
+                """
+            )
+            result = await db.execute(
+                sql,
+                {"q": tsquery, "tid": tenant_id, "lim": limit},
+            )
         rows = result.mappings().all()
         return [dict(r) for r in rows]
 
-    async def search_vector(
+    async def _search_vector_filtered(
         self,
         db: AsyncSession,
         embedding: list[float],
         tenant_id: uuid.UUID,
         limit: int,
+        visible_ids: set[uuid.UUID] | None,
     ) -> list[dict[str, Any]]:
-        """Semantic search on contacts.embedding."""
-        sql = text(
-            """
-            SELECT c.*, 1 - (c.embedding <=> cast(:emb AS vector)) AS score
-            FROM contacts c
-            WHERE c.tenant_id = :tid
-              AND c.deleted_at IS NULL
-              AND c.embedding IS NOT NULL
-            ORDER BY c.embedding <=> cast(:emb AS vector)
-            LIMIT :lim
-            """
-        )
-        result = await db.execute(
-            sql,
-            {"emb": str(embedding), "tid": tenant_id, "lim": limit},
-        )
+        """Semantic search on contacts.embedding, filtered by visible_ids.
+
+        If visible_ids is None, no visibility filter is applied (system admin).
+        """
+        if visible_ids is not None:
+            sql = text(
+                """
+                SELECT c.*, 1 - (c.embedding <=> cast(:emb AS vector)) AS score
+                FROM contacts c
+                WHERE c.tenant_id = :tid
+                  AND c.deleted_at IS NULL
+                  AND c.embedding IS NOT NULL
+                  AND c.id = ANY(:visible_ids)
+                ORDER BY c.embedding <=> cast(:emb AS vector)
+                LIMIT :lim
+                """
+            )
+            result = await db.execute(
+                sql,
+                {
+                    "emb": str(embedding),
+                    "tid": tenant_id,
+                    "lim": limit,
+                    "visible_ids": list(visible_ids),
+                },
+            )
+        else:
+            sql = text(
+                """
+                SELECT c.*, 1 - (c.embedding <=> cast(:emb AS vector)) AS score
+                FROM contacts c
+                WHERE c.tenant_id = :tid
+                  AND c.deleted_at IS NULL
+                  AND c.embedding IS NOT NULL
+                ORDER BY c.embedding <=> cast(:emb AS vector)
+                LIMIT :lim
+                """
+            )
+            result = await db.execute(
+                sql,
+                {"emb": str(embedding), "tid": tenant_id, "lim": limit},
+            )
         rows = result.mappings().all()
         return [dict(r) for r in rows]
 
     async def get_embedding_text(
-        self,
-        db: AsyncSession,
-        entity_id: uuid.UUID,
-        tenant_id: uuid.UUID,
+        self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID
     ) -> str:
         """Get text for embedding generation."""
         sql = text(

app/plugins/builtins/unified_search/routes.py

@@ -51,17 +51,20 @@ async def search(
     """Perform hybrid search with KI query understanding."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
     user_id = uuid.UUID(current_user["user_id"])
+    is_system_admin = current_user.get("is_system_admin", False)
 
     # KI query understanding
     query_analysis = await llm_analyze_query(req.query, db=db, tenant_id=tenant_id)
 
-    # Hybrid search
+    # Hybrid search with visibility filtering
     results = await hybrid_search(
         db=db,
         query_analysis=query_analysis,
         tenant_id=tenant_id,
         entity_types=req.entity_types,
         limit=req.limit,
+        user_id=user_id,
+        is_system_admin=is_system_admin,
     )
 
     # Resolve user permissions for field-level RBAC

app/plugins/builtins/unified_search/search_engine.py

@@ -68,11 +68,15 @@ async def hybrid_search(
     tenant_id: uuid.UUID,
     entity_types: list[str] | None = None,
     limit: int = 20,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> list[dict[str, Any]]:
     """Perform hybrid search across all entity types.
 
     For each entity: FTS search + vector search + RRF fusion.
     Merge all results, sort by fused score, return top limit.
+
+    Passes user_id and is_system_admin to each provider for visibility filtering.
     """
     registry = get_search_registry()
     all_entity_types = entity_types or registry.get_entity_types()
@@ -108,13 +112,13 @@ async def hybrid_search(
         vec_results: list[dict[str, Any]] = []
 
         try:
-            fts_results = await provider.search_fts(db, tsquery, tenant_id, fetch_limit)
+            fts_results = await provider.search_fts(db, tsquery, tenant_id, fetch_limit, user_id=user_id, is_system_admin=is_system_admin)
         except Exception:
             logger.exception("FTS search failed for %s", entity_type)
 
         if query_embedding:
             try:
-                vec_results = await provider.search_vector(db, query_embedding, tenant_id, fetch_limit)
+                vec_results = await provider.search_vector(db, query_embedding, tenant_id, fetch_limit, user_id=user_id, is_system_admin=is_system_admin)
             except Exception:
                 logger.exception("Vector search failed for %s", entity_type)
 

app/services/custom_field_definition_service.py

@@ -0,0 +1,172 @@
+"""CustomFieldDefinition service — CRUD with tenant isolation and visibility filter."""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.visibility import apply_visibility_filter, check_single_entity_access
+from app.models.custom_field_definition import CustomFieldDefinition
+
+
+def _definition_to_dict(d: CustomFieldDefinition) -> dict[str, Any]:
+    """Serialize a CustomFieldDefinition ORM object to dict."""
+    return {
+        "id": str(d.id),
+        "entity": d.entity,
+        "name": d.name,
+        "label": d.label,
+        "field_type": d.field_type,
+        "options": d.options,
+        "default_value": d.default_value,
+        "required": d.required,
+        "is_active": d.is_active,
+        "sort_order": d.sort_order,
+        "owner_id": str(d.owner_id) if d.owner_id else None,
+        "created_by": str(d.created_by) if d.created_by else None,
+        "updated_by": str(d.updated_by) if d.updated_by else None,
+    }
+
+
+async def list_custom_field_definitions(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    entity: str | None = None,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
+) -> list[dict[str, Any]]:
+    """List custom field definitions for a tenant, optionally filtered by entity."""
+    q = select(CustomFieldDefinition).where(
+        CustomFieldDefinition.tenant_id == tenant_id,
+        CustomFieldDefinition.is_active == True,  # noqa: E712
+    )
+    if entity:
+        q = q.where(CustomFieldDefinition.entity == entity)
+    if user_id and not is_system_admin:
+        q = await apply_visibility_filter(
+            db, q, "custom_field_definition", CustomFieldDefinition, user_id, tenant_id, is_system_admin
+        )
+    q = q.order_by(CustomFieldDefinition.sort_order, CustomFieldDefinition.name)
+    result = await db.execute(q)
+    return [_definition_to_dict(d) for d in result.scalars().all()]
+
+
+async def get_custom_field_definition(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    definition_id: uuid.UUID,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
+) -> dict[str, Any] | None:
+    """Get a single custom field definition by ID."""
+    q = select(CustomFieldDefinition).where(
+        CustomFieldDefinition.id == definition_id,
+        CustomFieldDefinition.tenant_id == tenant_id,
+    )
+    result = await db.execute(q)
+    definition = result.scalar_one_or_none()
+    if definition is None:
+        return None
+    if user_id and not is_system_admin:
+        has_access = await check_single_entity_access(
+            db, "custom_field_definition", definition.id, user_id, tenant_id, "read", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No access")
+    return _definition_to_dict(definition)
+
+
+async def create_custom_field_definition(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any]:
+    """Create a new custom field definition."""
+    definition = CustomFieldDefinition(
+        tenant_id=tenant_id,
+        entity=data["entity"],
+        name=data["name"],
+        label=data["label"],
+        field_type=data["field_type"],
+        options=data.get("options"),
+        default_value=data.get("default_value"),
+        required=data.get("required", False),
+        is_active=data.get("is_active", True),
+        sort_order=data.get("sort_order", 0),
+        created_by=user_id,
+        updated_by=user_id,
+        owner_id=user_id,
+    )
+    db.add(definition)
+    await db.flush()
+    await db.refresh(definition)
+    return _definition_to_dict(definition)
+
+
+async def update_custom_field_definition(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    definition_id: uuid.UUID,
+    data: dict[str, Any],
+    is_system_admin: bool = False,
+) -> dict[str, Any] | None:
+    """Update an existing custom field definition."""
+    q = select(CustomFieldDefinition).where(
+        CustomFieldDefinition.id == definition_id,
+        CustomFieldDefinition.tenant_id == tenant_id,
+    )
+    result = await db.execute(q)
+    definition = result.scalar_one_or_none()
+    if definition is None:
+        return None
+
+    if not is_system_admin:
+        has_access = await check_single_entity_access(
+            db, "custom_field_definition", definition.id, user_id, tenant_id, "write", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No access")
+
+    update_fields = ["label", "field_type", "options", "default_value", "required", "is_active", "sort_order"]
+    for field in update_fields:
+        if field in data:
+            setattr(definition, field, data[field])
+
+    definition.updated_by = user_id
+    await db.flush()
+    await db.refresh(definition)
+    return _definition_to_dict(definition)
+
+
+async def delete_custom_field_definition(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    definition_id: uuid.UUID,
+    is_system_admin: bool = False,
+) -> bool:
+    """Delete a custom field definition."""
+    q = select(CustomFieldDefinition).where(
+        CustomFieldDefinition.id == definition_id,
+        CustomFieldDefinition.tenant_id == tenant_id,
+    )
+    result = await db.execute(q)
+    definition = result.scalar_one_or_none()
+    if definition is None:
+        return False
+
+    if not is_system_admin:
+        has_access = await check_single_entity_access(
+            db, "custom_field_definition", definition.id, user_id, tenant_id, "admin", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No access")
+
+    await db.delete(definition)
+    await db.flush()
+    return True

app/services/custom_field_service.py

@@ -8,6 +8,7 @@ from typing import Any
 from sqlalchemy import select, delete
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.visibility import apply_visibility_filter, check_single_entity_access
 from app.models.custom_field_definition import CustomFieldDefinition
 
 
@@ -15,6 +16,8 @@ async def list_definitions(
     db: AsyncSession,
     tenant_id: uuid.UUID,
     entity: str | None = None,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> list[CustomFieldDefinition]:
     """List custom field definitions for a tenant, optionally filtered by entity."""
     stmt = select(CustomFieldDefinition).where(
@@ -23,6 +26,10 @@ async def list_definitions(
     )
     if entity:
         stmt = stmt.where(CustomFieldDefinition.entity == entity)
+    if user_id and not is_system_admin:
+        stmt = await apply_visibility_filter(
+            db, stmt, "custom_field_definition", CustomFieldDefinition, user_id, tenant_id, is_system_admin
+        )
     stmt = stmt.order_by(CustomFieldDefinition.sort_order, CustomFieldDefinition.name)
     result = await db.execute(stmt)
     return list(result.scalars().all())
@@ -32,6 +39,8 @@ async def get_definition(
     db: AsyncSession,
     tenant_id: uuid.UUID,
     definition_id: uuid.UUID,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> CustomFieldDefinition | None:
     """Get a single custom field definition by ID."""
     stmt = select(CustomFieldDefinition).where(
@@ -39,7 +48,16 @@ async def get_definition(
         CustomFieldDefinition.tenant_id == tenant_id,
     )
     result = await db.execute(stmt)
-    return result.scalar_one_or_none()
+    definition = result.scalar_one_or_none()
+    if definition is None:
+        return None
+    if user_id and not is_system_admin:
+        has_access = await check_single_entity_access(
+            db, "custom_field_definition", definition.id, user_id, tenant_id, "read", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No access")
+    return definition
 
 
 async def create_definition(
@@ -62,6 +80,7 @@ async def create_definition(
         sort_order=data.get("sort_order", 0),
         created_by=user_id,
         updated_by=user_id,
+        owner_id=user_id,
     )
     db.add(definition)
     await db.flush()
@@ -75,12 +94,20 @@ async def update_definition(
     definition_id: uuid.UUID,
     data: dict[str, Any],
     user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> CustomFieldDefinition | None:
     """Update an existing custom field definition."""
-    definition = await get_definition(db, tenant_id, definition_id)
+    definition = await get_definition(db, tenant_id, definition_id, user_id, is_system_admin)
     if definition is None:
         return None
 
+    if not is_system_admin:
+        has_access = await check_single_entity_access(
+            db, "custom_field_definition", definition.id, user_id, tenant_id, "write", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No access")
+
     update_fields = ["label", "field_type", "options", "default_value", "required", "is_active", "sort_order"]
     for field in update_fields:
         if field in data:
@@ -98,6 +125,8 @@ async def delete_definition(
     db: AsyncSession,
     tenant_id: uuid.UUID,
     definition_id: uuid.UUID,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> bool:
     """Delete a custom field definition."""
     stmt = select(CustomFieldDefinition).where(
@@ -108,6 +137,14 @@ async def delete_definition(
     definition = result.scalar_one_or_none()
     if definition is None:
         return False
+
+    if not is_system_admin:
+        has_access = await check_single_entity_access(
+            db, "custom_field_definition", definition.id, user_id, tenant_id, "admin", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No access")
+
     await db.delete(definition)
     await db.flush()
     return True

app/services/entity_history_service.py

@@ -9,6 +9,7 @@ from typing import Any
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.visibility import apply_visibility_filter, check_single_entity_access
 from app.models.entity_history import EntityHistory
 
 
@@ -33,6 +34,7 @@ async def record_history(
         snapshot_before=snapshot_before,
         snapshot_after=snapshot_after,
         changes=changes,
+        owner_id=user_id,
     )
     db.add(entry)
     await db.flush()
@@ -45,6 +47,8 @@ async def get_entity_history(
     entity_type: str,
     entity_id: uuid.UUID,
     limit: int = 50,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> list[EntityHistory]:
     """Get all history entries for an entity, newest first."""
     q = (
@@ -57,6 +61,10 @@ async def get_entity_history(
         .order_by(EntityHistory.created_at.desc())
         .limit(limit)
     )
+    if user_id and not is_system_admin:
+        q = await apply_visibility_filter(
+            db, q, "entity_history", EntityHistory, user_id, tenant_id, is_system_admin
+        )
     result = await db.execute(q)
     return list(result.scalars().all())
 
@@ -65,6 +73,8 @@ async def get_history_entry(
     db: AsyncSession,
     tenant_id: uuid.UUID,
     history_id: uuid.UUID,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> EntityHistory | None:
     """Get a specific history entry by ID."""
     q = select(EntityHistory).where(
@@ -72,7 +82,16 @@ async def get_history_entry(
         EntityHistory.tenant_id == tenant_id,
     )
     result = await db.execute(q)
-    return result.scalar_one_or_none()
+    entry = result.scalar_one_or_none()
+    if entry is None:
+        return None
+    if user_id and not is_system_admin:
+        has_access = await check_single_entity_access(
+            db, "entity_history", entry.id, user_id, tenant_id, "read", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No access")
+    return entry
 
 
 async def restore_from_history(
@@ -80,6 +99,7 @@ async def restore_from_history(
     tenant_id: uuid.UUID,
     history_id: uuid.UUID,
     user_id: uuid.UUID,
+    is_system_admin: bool = False,
 ) -> dict[str, Any]:
     """Restore an entity to a previous snapshot state.
 
@@ -89,7 +109,7 @@ async def restore_from_history(
 
     Returns the restored data dict.
     """
-    entry = await get_history_entry(db, tenant_id, history_id)
+    entry = await get_history_entry(db, tenant_id, history_id, user_id, is_system_admin)
     if entry is None:
         raise ValueError("History entry not found")
 
@@ -175,6 +195,7 @@ async def undo_last_action(
     user_id: uuid.UUID,
     entity_type: str,
     entity_id: uuid.UUID,
+    is_system_admin: bool = False,
 ) -> dict[str, Any]:
     """Undo the most recent action for an entity.
 
@@ -191,9 +212,13 @@ async def undo_last_action(
         .order_by(EntityHistory.created_at.desc())
         .limit(1)
     )
+    if not is_system_admin:
+        q = await apply_visibility_filter(
+            db, q, "entity_history", EntityHistory, user_id, tenant_id, is_system_admin
+        )
     result = await db.execute(q)
     entry = result.scalar_one_or_none()
     if entry is None:
         raise ValueError("No history found for this entity")
 
-    return await restore_from_history(db, tenant_id, entry.id, user_id)
+    return await restore_from_history(db, tenant_id, entry.id, user_id, is_system_admin)