sprint2: visibility filter + contact service access checks + contacts route integration

RBAC_PROGRESS.md

@@ -1,74 +1,63 @@
 # RBAC Build Progress — LeoCRM
 
-## Letztes Update: 2026-07-29 01:28 CEST
+## Letztes Update: 2026-07-29 01:31 CEST
 
-## Sprint 1 — Fundament (14h)
+## Sprint 1 — Fundament (14h) ✅ VOLLSTÄNDIG
 
-### Erledigt ✅
-- [x] EntityPermission Model erstellt (`app/models/entity_permission.py`)
-  - Generische Tabelle für alle Entities
-  - principal_type: user | group | role | guest
-  - permission_level: none | read | write | admin | delete
-  - expires_at für zeitlich begrenzte Freigaben
-  - created_by für Audit-Trail
-  - Indexes auf entity, principal, tenant, expires_at
+### 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)
 
-- [x] OwnedMixin erstellt (`app/models/owned_mixin.py`)
-  - owner_id Feld (nullable, FK → users.id, ON DELETE SET NULL)
-  - Index auf owner_id
-  - Kann auf jedes Model angewendet werden
+## Sprint 2 — Row-Level Security (16h) 🔄 NÄCHSTER
 
-- [x] Migration 0049: entity_permissions Tabelle — ✅ Ausgeführt in Produktion
-- [x] Migration 0050: owner_id auf 15 Tabellen — ✅ Ausgeführt in Produktion
+### 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)
+- [ ] Copy/Duplicate Permission
+- [ ] EXISTS-Optimization für RLS
 
-- [x] OwnedMixin auf alle 13 Models angewendet:
-  - Contact, Address, Attachment, BankAccount, Workflow, Sequence
-  - SavedFilter, SavedView, Webhook, Notification, CustomFieldDefinition
-  - EntityHistory, AIConversation
-
-- [x] Universeller Permission Service (`app/services/entity_permission_service.py`, 648 Zeilen)
-  - list_permissions, create_permission, update_permission, delete_permission
-  - get_effective_access (Owner → User → Group → Role → None)
-  - get_visible_ids (alle sichtbaren Datensätze für User)
-  - batch_get_effective_access (Batch-Resolution für Listen)
-  - get_cached_visible_ids (Redis Cache, 5 Min TTL)
-  - check_entity_access (einfacher Check)
-  - cleanup_expired_permissions (Background Worker)
-
-- [x] Universelle Permission API (`app/routes/entity_permissions.py`, 151 Zeilen)
-  - GET /api/v1/permissions/{entity_type}/{entity_id}
-  - POST /api/v1/permissions/{entity_type}/{entity_id}
-  - PUT /api/v1/permissions/{entity_type}/{entity_id}/{permission_id}
-  - DELETE /api/v1/permissions/{entity_type}/{entity_id}/{permission_id}
-  - GET /api/v1/permissions/{entity_type}/{entity_id}/access
-  - GET /api/v1/permissions/registry
-
-- [x] Schema erstellt (`app/schemas/entity_permission.py`)
-- [x] Route in main.py registriert
-- [x] Alle Imports getestet — OK
-- [x] Container neu gestartet
-- [x] Git committed und gepusht (5afa1fa)
-
-### In Bearbeitung 🔄
-- [ ] PostgreSQL RLS Policies + set_user_context()
-- [ ] Rate Limiting auf Permission-Änderungen
-- [ ] Folder ACLs migrieren (Migration 0051)
-
-### Noch offen ⬜
-- [ ] Sprint 2-23 (siehe ENTERPRISE_RBAC_PLAN.md)
+## Sprint 3-23 ⬜
 
 ---
 
 ## Migrationen
 | # | Beschreibung | Status |
 |---|-------------|:---:|
-| 0048 | contact_folder_permissions Tabelle | ✅ Ausgeführt |
-| 0049 | entity_permissions Tabelle | ✅ Ausgeführt |
-| 0050 | owner_id auf 15 Tabellen | ✅ Ausgeführt |
-| 0051 | Folder ACLs → entity_permissions | ⬜ Geplant |
+| 0048 | contact_folder_permissions Tabelle | ✅ |
+| 0049 | entity_permissions Tabelle | ✅ |
+| 0050 | owner_id auf 15 Tabellen | ✅ |
+| 0051 | Folder ACLs → entity_permissions | ✅ |
+| 0052 | RLS Policies auf contacts | ✅ |
 
 ## Git Commits
 | Hash | Beschreibung |
 |------|-------------|
-| cc021cd | feat: folder permissions (ACLs) - share folders with users/groups |
-| 5afa1fa | sprint1: entity_permissions table + owned_mixin + universal permission service + API + migrations 0049+0050 |
+| cc021cd | feat: folder permissions (ACLs) |
+| 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 |
+
+## 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
+- Universelle Permission API unter /api/v1/permissions/*
+- Rate Limiting auf Permission-Änderungen
+- Folder ACLs in entity_permissions migriert

app/core/visibility.py

@@ -0,0 +1,237 @@
+"""Visibility filter helper — applies row-level security to SQLAlchemy queries.
+
+This is the core function that ALL routes use to filter queries based on
+the current user's permissions. It works alongside PostgreSQL RLS as a
+Defense-in-Depth layer.
+
+Usage:
+    from app.core.visibility import apply_visibility_filter
+    
+    @router.get("/contacts")
+    async def list_contacts(db, current_user):
+        query = select(Contact).where(Contact.tenant_id == tenant_id)
+        query = await apply_visibility_filter(
+            db, query, "contact", Contact, user_id, tenant_id
+        )
+        result = await db.execute(query)
+        ...
+
+Architecture:
+- System admin → no filter (sees everything)
+- Non-admin → filter by: owner_id = user OR owner_id IS NULL OR shared via entity_permissions
+- Uses EXISTS subquery for performance (better than IN)
+- Works with any entity type that has owner_id column
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy import and_, exists, or_, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import DeclarativeBase
+
+from app.models.entity_permission import EntityPermission
+from app.models.group import UserGroup
+from app.models.user import User, UserTenant
+
+logger = logging.getLogger(__name__)
+
+# Permission rank for comparison
+_PERM_RANK = {"none": 0, "read": 1, "write": 2, "admin": 3, "delete": 4, "owner": 5}
+
+
+def _rank(level: str) -> int:
+    return _PERM_RANK.get(level, 0)
+
+
+async def _get_user_principals(
+    db: AsyncSession,
+    user_id: uuid.UUID,
+    tenant_id: uuid.UUID,
+) -> tuple[list[uuid.UUID], uuid.UUID | None]:
+    """Get user's group IDs and role ID for permission resolution."""
+    groups_q = await db.execute(
+        select(UserGroup.group_id)
+        .where(UserGroup.user_id == user_id)
+        .where(UserGroup.tenant_id == tenant_id)
+    )
+    group_ids = [row[0] for row in groups_q]
+
+    role_q = await db.execute(
+        select(UserTenant.role_id)
+        .where(UserTenant.user_id == user_id)
+        .where(UserTenant.tenant_id == tenant_id)
+    )
+    role_id = role_q.scalar_one_or_none()
+
+    return group_ids, role_id
+
+
+async def apply_visibility_filter(
+    db: AsyncSession,
+    query: Any,
+    entity_type: str,
+    model: type[DeclarativeBase],
+    user_id: uuid.UUID,
+    tenant_id: uuid.UUID,
+    is_system_admin: bool = False,
+) -> Any:
+    """Apply row-level visibility filter to a SQLAlchemy query.
+
+    This function modifies the query to only return rows that the user
+    is allowed to see based on:
+    1. System admin → no filter (sees everything)
+    2. Owner → rows where owner_id = user_id
+    3. Tenant-owned → rows where owner_id IS NULL
+    4. Shared → rows with entity_permissions entry for this user/group/role
+
+    Args:
+        db: Database session
+        query: SQLAlchemy select() query to filter
+        entity_type: Entity type string (e.g. 'contact', 'address')
+        model: SQLAlchemy model class (must have owner_id column)
+        user_id: Current user's UUID
+        tenant_id: Current tenant's UUID
+        is_system_admin: Whether user is system admin
+
+    Returns:
+        Modified query with visibility filter applied
+    """
+    if is_system_admin:
+        return query  # System admin sees everything
+
+    # Get user's groups and role
+    group_ids, role_id = await _get_user_principals(db, user_id, tenant_id)
+
+    # Build principal conditions for entity_permissions EXISTS subquery
+    principal_conditions = [
+        and_(
+            EntityPermission.principal_type == "user",
+            EntityPermission.principal_id == user_id,
+        ),
+    ]
+    if group_ids:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "group",
+                EntityPermission.principal_id.in_(group_ids),
+            )
+        )
+    if role_id:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "role",
+                EntityPermission.principal_id == role_id,
+            )
+        )
+
+    # Build EXISTS subquery for shared entities
+    # Uses EXISTS instead of IN for better PostgreSQL optimization
+    shared_exists = (
+        select(EntityPermission.id)
+        .where(EntityPermission.entity_type == entity_type)
+        .where(EntityPermission.entity_id == model.id)
+        .where(EntityPermission.tenant_id == tenant_id)
+        .where(EntityPermission.permission_level != "none")
+        .where(
+            or_(
+                EntityPermission.expires_at.is_(None),
+                EntityPermission.expires_at > text("NOW()"),
+            )
+        )
+        .where(or_(*principal_conditions))
+        .exists()
+    )
+
+    # Apply filter: owner OR tenant-owned OR shared
+    visibility_condition = or_(
+        model.owner_id == user_id,  # Own entities
+        model.owner_id.is_(None),   # Tenant-owned entities
+        shared_exists,              # Shared via entity_permissions
+    )
+
+    return query.where(visibility_condition)
+
+
+async def check_single_entity_access(
+    db: AsyncSession,
+    entity_type: str,
+    entity_id: uuid.UUID,
+    user_id: uuid.UUID,
+    tenant_id: uuid.UUID,
+    required_level: str = "read",
+    is_system_admin: bool = False,
+) -> bool:
+    """Check if user has at least the required access level on a single entity.
+
+    Used for GET/PUT/DELETE on individual entities.
+    """
+    if is_system_admin:
+        return True
+
+    from app.services.entity_permission_service import get_effective_access
+    access = await get_effective_access(
+        db, tenant_id, user_id, entity_type, entity_id
+    )
+    return _rank(access) >= _rank(required_level)
+
+
+async def filter_response_fields(
+    data: dict[str, Any],
+    field_permissions: dict[str, dict[str, str]],
+    module: str,
+    is_system_admin: bool = False,
+) -> dict[str, Any]:
+    """Filter response fields based on field-level permissions.
+
+    Removes fields marked as 'hidden', keeps others.
+    This is a convenience wrapper that can be used in any route.
+    """
+    if is_system_admin:
+        return data
+
+    module_perms = field_permissions.get(module, {})
+    if not module_perms:
+        return data
+
+    return {
+        key: value
+        for key, value in data.items()
+        if module_perms.get(key, "read") != "hidden"
+    }
+
+
+async def apply_visibility_filter_cached(
+    db: AsyncSession,
+    redis: Any,
+    query: Any,
+    entity_type: str,
+    model: type[DeclarativeBase],
+    user_id: uuid.UUID,
+    tenant_id: uuid.UUID,
+    is_system_admin: bool = False,
+) -> Any:
+    """Apply visibility filter using cached visible IDs from Redis.
+
+    This is an alternative to apply_visibility_filter() that uses
+    pre-computed visible IDs from Redis cache for better performance.
+
+    Use this for list queries where you need maximum performance.
+    """
+    if is_system_admin:
+        return query
+
+    from app.services.entity_permission_service import get_cached_visible_ids
+
+    visible_ids, _ = await get_cached_visible_ids(
+        db, redis, tenant_id, user_id, entity_type
+    )
+
+    if not visible_ids:
+        # No visible entities — return empty result
+        return query.where(text("1 = 0"))
+
+    return query.where(model.id.in_(list(visible_ids)))

app/routes/contacts.py

@@ -69,12 +69,16 @@ async def list_contacts(
 ):
     """List contacts with pagination, FTS search, type/folder filter, sorting."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    is_admin = current_user.get("is_system_admin", False)
     return await contact_service.list_contacts(
         db, tenant_id,
         page=page, page_size=page_size, search=search,
         contact_type=type, folder_id=folder_id,
         sort_by=sort_by, sort_order=sort_order,
         resolved_perms=current_user,
+        user_id=user_id,
+        is_system_admin=is_admin,
     )
 
 
@@ -88,7 +92,12 @@ async def export_contacts(
 ):
     """Stream contacts as CSV."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
-    csv_data = await contact_service.export_contacts_csv(db, tenant_id, contact_type=type, search=search)
+    user_id = uuid.UUID(current_user["user_id"])
+    is_admin = current_user.get("is_system_admin", False)
+    csv_data = await contact_service.export_contacts_csv(
+        db, tenant_id, contact_type=type, search=search,
+        user_id=user_id, is_system_admin=is_admin,
+    )
     return StreamingResponse(
         io.StringIO(csv_data),
         media_type="text/csv",
@@ -132,10 +141,14 @@ async def get_contact(
 ):
     """Get a single contact with contact_persons."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    is_admin = current_user.get("is_system_admin", False)
     try:
-        return await contact_service.get_contact(db, tenant_id, contact_id)
+        return await contact_service.get_contact(db, tenant_id, contact_id, user_id=user_id, is_system_admin=is_admin)
     except ValueError as e:
         raise HTTPException(status_code=404, detail=str(e))
+    except PermissionError as e:
+        raise HTTPException(status_code=403, detail=str(e))
 
 
 @router.put("/{contact_id}")

app/services/contact_service.py

@@ -145,13 +145,26 @@ async def list_contacts(
     sort_by: str = "displayname",
     sort_order: str = "asc",
     resolved_perms: dict | None = None,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> dict:
-    """List contacts with pagination, FTS search, type/folder filter, sorting."""
+    """List contacts with pagination, FTS search, type/folder filter, sorting.
+
+    Applies row-level visibility filter based on ownership and entity_permissions.
+    """
+    from app.core.visibility import apply_visibility_filter
+
     base = select(Contact).where(
         Contact.tenant_id == tenant_id,
         Contact.deleted_at.is_(None),
     )
 
+    # Apply row-level visibility filter
+    if user_id and not is_system_admin:
+        base = await apply_visibility_filter(
+            db, base, "contact", Contact, user_id, tenant_id, is_system_admin
+        )
+
     if contact_type:
         base = base.where(Contact.type == contact_type)
 
@@ -191,8 +204,9 @@ async def list_contacts(
     }
 
 
-async def get_contact(db: AsyncSession, tenant_id: uuid.UUID, contact_id: str) -> dict:
-    """Get a single contact with contact_persons."""
+async def get_contact(db: AsyncSession, tenant_id: uuid.UUID, contact_id: str,
+                     user_id: uuid.UUID | None = None, is_system_admin: bool = False) -> dict:
+    """Get a single contact with contact_persons. Checks row-level access."""
     q = (
         select(Contact)
         .options(selectinload(Contact.contact_persons))
@@ -206,6 +220,16 @@ async def get_contact(db: AsyncSession, tenant_id: uuid.UUID, contact_id: str) -
     contact = result.scalar_one_or_none()
     if not contact:
         raise ValueError("Contact not found")
+    
+    # Check row-level access
+    if user_id and not is_system_admin:
+        from app.core.visibility import check_single_entity_access
+        has_access = await check_single_entity_access(
+            db, "contact", contact.id, user_id, tenant_id, "read", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No access to this contact")
+    
     return _serialize_contact_detail(contact)
 
 
@@ -222,6 +246,7 @@ async def create_contact(
         tenant_id=tenant_id,
         created_by=user_id,
         updated_by=user_id,
+        owner_id=user_id,
         **{k: v for k, v in data.items() if hasattr(Contact, k)},
     )
     db.add(contact)
@@ -277,9 +302,10 @@ async def create_contact(
 
 
 async def update_contact(
-    db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, contact_id: str, data: dict
+    db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, contact_id: str, data: dict,
+    is_system_admin: bool = False,
 ) -> dict:
-    """Update a contact."""
+    """Update a contact. Checks row-level write access."""
     # Expire all cached objects to ensure fresh data with selectinload
     db.expire_all()
     q = (
@@ -296,6 +322,15 @@ async def update_contact(
     if not contact:
         raise ValueError("Contact not found")
 
+    # Check row-level write access
+    if not is_system_admin:
+        from app.core.visibility import check_single_entity_access
+        has_access = await check_single_entity_access(
+            db, "contact", contact.id, user_id, tenant_id, "write", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No write access to this contact")
+
     # Hook: contact.before_update
     await do_action("contact.before_update", data, db=db, tenant_id=tenant_id, user_id=user_id, contact_id=contact_id)
 
@@ -356,9 +391,10 @@ async def update_contact(
 
 
 async def delete_contact(
-    db: AsyncSession, tenant_id: uuid.UUID, contact_id: str, user_id: uuid.UUID | None = None
+    db: AsyncSession, tenant_id: uuid.UUID, contact_id: str, user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> None:
-    """Soft-delete a contact."""
+    """Soft-delete a contact. Checks row-level admin access."""
     q = select(Contact).where(
         Contact.id == uuid.UUID(contact_id),
         Contact.tenant_id == tenant_id,
@@ -369,6 +405,15 @@ async def delete_contact(
     if not contact:
         raise ValueError("Contact not found")
 
+    # Check row-level admin access
+    if not is_system_admin:
+        from app.core.visibility import check_single_entity_access
+        has_access = await check_single_entity_access(
+            db, "contact", contact.id, user_id, tenant_id, "admin", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError("No admin access to this contact")
+
     # Hook: contact.before_delete
     await do_action("contact.before_delete", db=db, tenant_id=tenant_id, contact_id=contact_id, user_id=user_id)
 
@@ -491,9 +536,12 @@ async def delete_contact_person(
 
 
 async def export_contacts_csv(
-    db: AsyncSession, tenant_id: uuid.UUID, contact_type: str | None = None, search: str | None = None
+    db: AsyncSession, tenant_id: uuid.UUID, contact_type: str | None = None, search: str | None = None,
+    user_id: uuid.UUID | None = None, is_system_admin: bool = False,
 ) -> str:
-    """Export contacts as CSV string."""
+    """Export contacts as CSV string. Only exports visible contacts."""
+    from app.core.visibility import apply_visibility_filter
+
     base = select(Contact).where(
         Contact.tenant_id == tenant_id,
         Contact.deleted_at.is_(None),
@@ -502,6 +550,13 @@ async def export_contacts_csv(
         base = base.where(Contact.type == contact_type)
     if search:
         base = base.where(Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search)))
+    
+    # Apply visibility filter
+    if user_id and not is_system_admin:
+        base = await apply_visibility_filter(
+            db, base, "contact", Contact, user_id, tenant_id, is_system_admin
+        )
+    
     base = base.order_by(Contact.displayname)
 
     result = await db.execute(base)