leocrm/app/core/visibility.py

"""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)))