"""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 # Defense-in-Depth: Always filter by tenant_id first (P0.4 fix) # This ensures cross-tenant data is never returned even if RLS is bypassed if hasattr(model, 'tenant_id'): query = query.where(model.tenant_id == tenant_id) # 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)))