leocrm/app/services/entity_permission_service.py

"""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 logging
import uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import String, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.audit import log_audit
from app.core.notifications import post_system_message
from app.models.address import Address
from app.models.attachment import Attachment
from app.models.bank_account import BankAccount
from app.models.contact_folder import ContactFolder
from app.models.custom_field_definition import CustomFieldDefinition
from app.models.entity_permission import EntityPermission
from app.models.group import Group, UserGroup
from app.models.notification import Notification
from app.models.role import Role
from app.models.saved_filter import SavedFilter
from app.models.saved_view import SavedView
from app.models.sequence import Sequence
from app.models.user import User
from app.models.webhook import Webhook
from app.models.workflow import Workflow

# Import cache helpers used by CRUD operations
from app.services.permission_cache import _invalidate_user_cache

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] = {
    # Core models only — plugin models (incl. contacts: contact/contacts/company)
    # are registered dynamically via plugin.get_entity_models() at activation
    # time (P0-3 fix). The contacts entries were duplicated here historically
    # — ContactsPlugin.get_entity_models() is the single source of truth.
    "address": Address,
    "attachment": Attachment,
    "bank_account": BankAccount,
    "workflow": Workflow,
    "sequence": Sequence,
    "saved_filter": SavedFilter,
    "saved_view": SavedView,
    "webhook": Webhook,
    "notification": Notification,
    "custom_field_definition": CustomFieldDefinition,
    "contact_folder": ContactFolder,
}

# W4b: Tracks which plugin registered which entity_type — used to derive
# the correct module permission (e.g. contacts:read for contacts entities).
ENTITY_PLUGIN_OWNERS: dict[str, str] = {}


def get_entity_read_permission(entity_type: str) -> str:
    """Derive the module read permission for an entity type.

    W4b: Saved views/filters must respect the owning plugin's permission
    instead of a hardcoded contacts:read. Falls back to contacts:read for
    unknown entities (backward compat, pre-plugin behavior).
    """
    owner = ENTITY_PLUGIN_OWNERS.get(entity_type)
    if owner:
        return f"{owner}:read"
    # Core entities: derive from module name (e.g. workflows → workflows:read)
    module = entity_type.rstrip("s")
    candidates = [k for k in _core_module_keys(module, "read")]
    return candidates[0] if candidates else "contacts:read"


def _core_module_keys(module: str, action: str) -> list[str]:
    """Find a core permission key matching module+action (lazy import safe)."""
    from app.core.permission_registry import CORE_PERMISSIONS

    return [
        p["key"]
        for p in CORE_PERMISSIONS
        if p.get("module") == module and p["key"].endswith(f":{action}")
    ]

# 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 are registered dynamically via plugin.get_entity_models()
# at activation time in main.py:lifespan(). No hardcoded plugin imports here.


def register_entity_model(
    entity_type: str,
    model_class: type,
    plugin_name: str | None = None,
) -> None:
    """Register an entity model dynamically (called during plugin activation)."""
    ENTITY_MODELS[entity_type] = model_class
    if plugin_name:
        ENTITY_PLUGIN_OWNERS[entity_type] = plugin_name


def unregister_entity_model(entity_type: str) -> None:
    """Unregister an entity model (called during plugin deactivation)."""
    ENTITY_MODELS.pop(entity_type, None)


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  # noqa: E402


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 post_system_message(
                db, tenant_id, principal_uuid,
                message_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 post_system_message(
            db, tenant_id, principal_uuid,
            message_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 post_system_message(
            db, tenant_id, old_principal_id,
            message_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 (intentional re-exports used by other modules)
from app.services.permission_cache import (  # noqa: E402
    get_cached_visible_ids as get_cached_visible_ids,
)
from app.services.permission_cache import (  # noqa: E402
    invalidate_all_user_entity_cache as invalidate_all_user_entity_cache,
)
from app.services.permission_resolver import (  # noqa: E402
    batch_get_effective_access as batch_get_effective_access,
)
from app.services.permission_resolver import (  # noqa: E402
    check_entity_access as check_entity_access,
)
from app.services.permission_resolver import (  # noqa: E402
    get_effective_access as get_effective_access,
)
from app.services.permission_resolver import (  # noqa: E402
    get_visible_ids as get_visible_ids,
)