"""Universal entity permission service — ACL management for ALL entities. This service handles: - CRUD for entity_permissions - Effective access resolution (owner → user → group → role → guest) - Batch resolution for list queries - Redis caching with bitmap optimization - Permission expiration checks - Audit logging for permission changes Resolution, caching, and audit logic have been extracted into focused modules: - permission_resolver.py: get_effective_access, get_visible_ids, batch_get_effective_access, check_entity_access - permission_cache.py: _invalidate_user_cache, get_cached_visible_ids, invalidate_all_user_entity_cache - permission_audit.py: _log_permission_grant, _log_permission_update, _log_permission_revoke """ from __future__ import annotations import json import logging import uuid from datetime import datetime, UTC from typing import Any import redis.asyncio as aioredis from sqlalchemy import and_, func, or_, select, text from sqlalchemy.ext.asyncio import AsyncSession from app.core.audit import log_audit from app.core.notifications import create_notification from app.models.entity_permission import EntityPermission from app.models.group import Group, UserGroup from app.models.role import Role from app.models.user import User, UserTenant from app.models.contact import Contact from app.models.address import Address from app.models.attachment import Attachment from app.models.bank_account import BankAccount from app.models.workflow import Workflow from app.models.sequence import Sequence from app.models.saved_filter import SavedFilter from app.models.saved_view import SavedView from app.models.webhook import Webhook from app.models.custom_field_definition import CustomFieldDefinition from app.models.contact_folder import ContactFolder # Import cache helpers used by CRUD operations from app.services.permission_cache import _invalidate_user_cache, CACHE_TTL, CACHE_PREFIX logger = logging.getLogger(__name__) # ── Entity Model Registry ──────────────────────────────────────────────────── # Maps entity_type string to SQLAlchemy model class. # This replaces insecure text(f"SELECT ... FROM {entity_type}s") queries # with safe SQLAlchemy model-based queries (prevents SQL injection). ENTITY_MODELS: dict[str, type] = { "contact": Contact, "contacts": Contact, "company": Contact, "address": Address, "attachment": Attachment, "bank_account": BankAccount, "workflow": Workflow, "sequence": Sequence, "saved_filter": SavedFilter, "saved_view": SavedView, "webhook": Webhook, "custom_field_definition": CustomFieldDefinition, "contact_folder": ContactFolder, } # Core models with OwnedMixin (Phase 2 additions) try: from app.models.entity_attachment import EntityAttachment ENTITY_MODELS["entity_attachment"] = EntityAttachment except ImportError: pass try: from app.models.entity_history import EntityHistory ENTITY_MODELS["entity_history"] = EntityHistory except ImportError: pass # Plugin models if available try: from app.plugins.builtins.dms.models import File as DmsFile ENTITY_MODELS["file"] = DmsFile except ImportError: pass try: from app.plugins.builtins.dms.models import Folder as DmsFolder ENTITY_MODELS["folder"] = DmsFolder except ImportError: pass try: from app.plugins.builtins.calendar.models import CalendarEntry ENTITY_MODELS["calendar_event"] = CalendarEntry except ImportError: pass try: from app.plugins.builtins.calendar.models import Calendar ENTITY_MODELS["calendar"] = Calendar except ImportError: pass try: from app.plugins.builtins.calendar.models import Subtask ENTITY_MODELS["subtask"] = Subtask except ImportError: pass try: from app.plugins.builtins.tasks.models import Task ENTITY_MODELS["task"] = Task except ImportError: pass try: from app.plugins.builtins.mail.models import MailAccount ENTITY_MODELS["mailbox"] = MailAccount ENTITY_MODELS["mail_account"] = MailAccount except ImportError: pass # Additional plugin models with OwnedMixin try: from app.plugins.builtins.mail.models import MailMessage ENTITY_MODELS["mail_message"] = MailMessage except ImportError: pass try: from app.plugins.builtins.kommunikation.models import CommConversation ENTITY_MODELS["comm_conversation"] = CommConversation except ImportError: pass try: from app.plugins.builtins.tags.models import Tag ENTITY_MODELS["tag"] = Tag except ImportError: pass try: from app.plugins.builtins.agent_memory.models import AgentMemory ENTITY_MODELS["agent_memory"] = AgentMemory except ImportError: pass try: from app.plugins.builtins.graph_rag.models import EntityRelationship ENTITY_MODELS["entity_relationship"] = EntityRelationship except ImportError: pass try: from app.plugins.builtins.report_generator.models import ReportTemplate, ReportInstance ENTITY_MODELS["report_template"] = ReportTemplate ENTITY_MODELS["report_instance"] = ReportInstance except ImportError: pass try: from app.plugins.builtins.entity_links.models import EntityLink ENTITY_MODELS["entity_link"] = EntityLink except ImportError: pass try: from app.plugins.builtins.kommunikation.models import CommConversation as CommConv ENTITY_MODELS["comm_conversation"] = CommConv except ImportError: pass try: from app.plugins.builtins.ai_proactive.models import ProactiveSuggestion ENTITY_MODELS["proactive_suggestion"] = ProactiveSuggestion except ImportError: pass try: from app.plugins.builtins.ai_assistant.models import AIAgent, AIChatSession ENTITY_MODELS["ai_agent"] = AIAgent ENTITY_MODELS["ai_chat_session"] = AIChatSession except ImportError: pass try: from app.plugins.builtins.permissions.models import ShareLink ENTITY_MODELS["share_link"] = ShareLink except ImportError: pass try: from app.plugins.builtins.automation.models import AgentDefinition, AutomationDefinition ENTITY_MODELS["agent_definition"] = AgentDefinition ENTITY_MODELS["automation_definition"] = AutomationDefinition except ImportError: pass try: from app.plugins.builtins.mcp_client.models import McpServerConfig ENTITY_MODELS["mcp_server_config"] = McpServerConfig except ImportError: pass def _get_entity_model(entity_type: str) -> type: """Get SQLAlchemy model class for entity_type, or raise ValueError.""" model = ENTITY_MODELS.get(entity_type) if model is None: raise ValueError(f"Unknown entity type: {entity_type}") return model from app.core.permissions import PERM_RANK as _PERM_RANK def _rank(level: str) -> int: return _PERM_RANK.get(level, 0) def _serialize_permission(p: EntityPermission, principal_name: str | None = None) -> dict: return { "id": str(p.id), "entity_type": p.entity_type, "entity_id": str(p.entity_id), "principal_type": p.principal_type, "principal_id": str(p.principal_id), "principal_name": principal_name, "permission_level": p.permission_level, "expires_at": p.expires_at.isoformat() if p.expires_at else None, "created_by": str(p.created_by) if p.created_by else None, "created_at": p.created_at.isoformat() if p.created_at else None, } async def _load_principal_names( db: AsyncSession, perms: list[EntityPermission] ) -> dict[uuid.UUID, str]: """Batch-load names for all principals in a permission list.""" user_ids = [p.principal_id for p in perms if p.principal_type == "user"] group_ids = [p.principal_id for p in perms if p.principal_type == "group"] role_ids = [p.principal_id for p in perms if p.principal_type == "role"] names: dict[uuid.UUID, str] = {} if user_ids: result = await db.execute(select(User.id, User.name).where(User.id.in_(user_ids))) names.update({row[0]: row[1] for row in result}) if group_ids: result = await db.execute(select(Group.id, Group.name).where(Group.id.in_(group_ids))) names.update({row[0]: row[1] for row in result}) if role_ids: result = await db.execute(select(Role.id, Role.name).where(Role.id.in_(role_ids))) names.update({row[0]: row[1] for row in result}) return names async def list_permissions( db: AsyncSession, tenant_id: uuid.UUID, entity_type: str, entity_id: str ) -> list[dict]: """List all permission entries for a specific entity.""" entity_uuid = uuid.UUID(entity_id) result = await db.execute( select(EntityPermission) .where(EntityPermission.entity_type == entity_type) .where(EntityPermission.entity_id == entity_uuid) .where(EntityPermission.tenant_id == tenant_id) .order_by(EntityPermission.created_at) ) perms = result.scalars().all() names = await _load_principal_names(db, perms) return [_serialize_permission(p, names.get(p.principal_id)) for p in perms] async def create_permission( db: AsyncSession, tenant_id: uuid.UUID, entity_type: str, entity_id: str, principal_type: str, principal_id: str, permission_level: str, expires_at: datetime | None = None, created_by: uuid.UUID | None = None, redis: Any = None, ) -> dict: """Create or update a permission entry (upsert).""" entity_uuid = uuid.UUID(entity_id) principal_uuid = uuid.UUID(principal_id) # Check for existing entry (upsert) existing_q = await db.execute( select(EntityPermission) .where(EntityPermission.entity_type == entity_type) .where(EntityPermission.entity_id == entity_uuid) .where(EntityPermission.principal_type == principal_type) .where(EntityPermission.principal_id == principal_uuid) .where(EntityPermission.tenant_id == tenant_id) ) existing = existing_q.scalar_one_or_none() if existing: existing.permission_level = permission_level existing.expires_at = expires_at await db.flush() await db.refresh(existing) names = await _load_principal_names(db, [existing]) # Audit log for permission update await log_audit( db, tenant_id, created_by, action='permission_update', entity_type='entity_permission', entity_id=existing.id, changes={'permission_level': permission_level, 'principal_type': principal_type, 'principal_id': principal_id} ) # Notify user if direct permission if principal_type == 'user': await create_notification( db, tenant_id, principal_uuid, type='permission_granted', title='Neue Berechtigung', body=f'{entity_type} wurde mit dir geteilt', entity_type=entity_type, entity_id=entity_uuid, ) # Invalidate cache for upsert if principal_type == "user": await _invalidate_user_cache(redis, tenant_id, principal_uuid, entity_type) elif principal_type == "group": members_q2 = await db.execute( select(UserGroup.user_id) .where(UserGroup.group_id == principal_uuid) .where(UserGroup.tenant_id == tenant_id) ) for (uid2,) in members_q2: await _invalidate_user_cache(redis, tenant_id, uid2, entity_type) return _serialize_permission(existing, names.get(existing.principal_id)) perm = EntityPermission( tenant_id=tenant_id, entity_type=entity_type, entity_id=entity_uuid, principal_type=principal_type, principal_id=principal_uuid, permission_level=permission_level, expires_at=expires_at, created_by=created_by, ) db.add(perm) await db.flush() await db.refresh(perm) # Invalidate cache for this principal if principal_type == "user": await _invalidate_user_cache(redis, tenant_id, principal_uuid, entity_type) elif principal_type == "group": # Invalidate for all group members members_q = await db.execute( select(UserGroup.user_id) .where(UserGroup.group_id == principal_uuid) .where(UserGroup.tenant_id == tenant_id) ) for (uid,) in members_q: await _invalidate_user_cache(redis, tenant_id, uid, entity_type) # Audit log for new permission await log_audit( db, tenant_id, created_by, action='permission_grant', entity_type='entity_permission', entity_id=perm.id, changes={'permission_level': permission_level, 'principal_type': principal_type, 'principal_id': principal_id} ) # Notify user if direct permission if principal_type == 'user': await create_notification( db, tenant_id, principal_uuid, type='permission_granted', title='Neue Berechtigung', body=f'{entity_type} wurde mit dir geteilt', entity_type=entity_type, entity_id=entity_uuid, ) names = await _load_principal_names(db, [perm]) return _serialize_permission(perm, names.get(perm.principal_id)) async def update_permission( db: AsyncSession, tenant_id: uuid.UUID, permission_id: str, permission_level: str, expires_at: datetime | None = None, redis: Any = None, ) -> dict: """Update an existing permission entry.""" perm_uuid = uuid.UUID(permission_id) result = await db.execute( select(EntityPermission) .where(EntityPermission.id == perm_uuid) .where(EntityPermission.tenant_id == tenant_id) ) perm = result.scalar_one_or_none() if not perm: raise ValueError("Permission not found") old_principal_type = perm.principal_type old_principal_id = perm.principal_id old_entity_type = perm.entity_type perm.permission_level = permission_level if expires_at is not None: perm.expires_at = expires_at await db.flush() await db.refresh(perm) # Invalidate cache if old_principal_type == "user": await _invalidate_user_cache(redis, tenant_id, old_principal_id, old_entity_type) elif old_principal_type == "group": members_q = await db.execute( select(UserGroup.user_id) .where(UserGroup.group_id == old_principal_id) .where(UserGroup.tenant_id == tenant_id) ) for (uid,) in members_q: await _invalidate_user_cache(redis, tenant_id, uid, old_entity_type) # Audit log for permission update await log_audit( db, tenant_id, None, action='permission_update', entity_type='entity_permission', entity_id=perm.id, changes={'permission_level': permission_level, 'principal_type': old_principal_type, 'principal_id': str(old_principal_id)} ) names = await _load_principal_names(db, [perm]) return _serialize_permission(perm, names.get(perm.principal_id)) async def delete_permission( db: AsyncSession, tenant_id: uuid.UUID, permission_id: str, redis: Any = None ) -> None: """Delete a permission entry.""" perm_uuid = uuid.UUID(permission_id) result = await db.execute( select(EntityPermission) .where(EntityPermission.id == perm_uuid) .where(EntityPermission.tenant_id == tenant_id) ) perm = result.scalar_one_or_none() if not perm: raise ValueError("Permission not found") old_principal_type = perm.principal_type old_principal_id = perm.principal_id old_entity_type = perm.entity_type old_entity_id = perm.entity_id # Audit log for permission revoke await log_audit( db, tenant_id, None, action='permission_revoke', entity_type='entity_permission', entity_id=perm.id, changes={'permission_level': perm.permission_level, 'principal_type': old_principal_type, 'principal_id': str(old_principal_id)} ) # Notify user if direct permission if old_principal_type == 'user': await create_notification( db, tenant_id, old_principal_id, type='permission_revoked', title='Berechtigung entfernt', body=f'{old_entity_type} wurde nicht mehr mit dir geteilt', entity_type=old_entity_type, entity_id=old_entity_id, ) await db.delete(perm) await db.flush() # Invalidate cache if old_principal_type == "user": await _invalidate_user_cache(redis, tenant_id, old_principal_id, old_entity_type) elif old_principal_type == "group": members_q = await db.execute( select(UserGroup.user_id) .where(UserGroup.group_id == old_principal_id) .where(UserGroup.tenant_id == tenant_id) ) for (uid,) in members_q: await _invalidate_user_cache(redis, tenant_id, uid, old_entity_type) async def list_all_permissions( db: AsyncSession, tenant_id: uuid.UUID ) -> list[dict]: """List ALL permission entries for a tenant (global view).""" result = await db.execute( select(EntityPermission) .where(EntityPermission.tenant_id == tenant_id) .order_by(EntityPermission.entity_type, EntityPermission.created_at) ) perms = result.scalars().all() names = await _load_principal_names(db, perms) return [_serialize_permission(p, names.get(p.principal_id)) for p in perms] async def get_permission_analytics( db: AsyncSession, tenant_id: uuid.UUID, ) -> dict: """Get permission analytics for a tenant. Returns: total_permissions: Total number of permission entries total_shared_entities: Number of unique entities with permissions permissions_by_level: Breakdown by permission level permissions_by_entity_type: Breakdown by entity type recent_changes: Last 10 permission changes """ from sqlalchemy import func as sa_func # Total permissions total_q = await db.execute( select(sa_func.count(EntityPermission.id)) .where(EntityPermission.tenant_id == tenant_id) ) total_permissions = total_q.scalar() or 0 # Total unique shared entities unique_q = await db.execute( select(sa_func.count(sa_func.distinct( EntityPermission.entity_type + ":" + EntityPermission.entity_id.cast(String) ))) .where(EntityPermission.tenant_id == tenant_id) ) total_shared_entities = unique_q.scalar() or 0 # Permissions by level level_q = await db.execute( select(EntityPermission.permission_level, sa_func.count(EntityPermission.id)) .where(EntityPermission.tenant_id == tenant_id) .group_by(EntityPermission.permission_level) ) permissions_by_level = {row[0]: row[1] for row in level_q} # Permissions by entity type type_q = await db.execute( select(EntityPermission.entity_type, sa_func.count(EntityPermission.id)) .where(EntityPermission.tenant_id == tenant_id) .group_by(EntityPermission.entity_type) ) permissions_by_entity_type = {row[0]: row[1] for row in type_q} # Recent changes (last 10) recent_q = await db.execute( select(EntityPermission) .where(EntityPermission.tenant_id == tenant_id) .order_by(EntityPermission.updated_at.desc()) .limit(10) ) recent = recent_q.scalars().all() recent_changes = [ { "id": str(p.id), "entity_type": p.entity_type, "entity_id": str(p.entity_id), "principal_type": p.principal_type, "principal_id": str(p.principal_id), "permission_level": p.permission_level, "updated_at": p.updated_at.isoformat() if p.updated_at else None, } for p in recent ] return { "total_permissions": total_permissions, "total_shared_entities": total_shared_entities, "permissions_by_level": permissions_by_level, "permissions_by_entity_type": permissions_by_entity_type, "recent_changes": recent_changes, } async def cleanup_expired_permissions(db: AsyncSession) -> int: """Delete all expired permission entries. Returns count deleted.""" now = datetime.now(UTC) result = await db.execute( select(EntityPermission).where(EntityPermission.expires_at < now) ) expired = result.scalars().all() count = len(expired) for perm in expired: await db.delete(perm) if count > 0: await db.flush() logger.info("Cleaned up %d expired entity permissions", count) return count # Backward compatibility re-exports from app.services.permission_resolver import ( # noqa: E402 get_effective_access, get_visible_ids, batch_get_effective_access, check_entity_access, ) from app.services.permission_cache import ( # noqa: E402 get_cached_visible_ids, invalidate_all_user_entity_cache, )