leocrm/app/plugins/builtins/kommunikation/rbac.py

"""Chat-internal RBAC logic — two-level permission checks."""

from __future__ import annotations

import logging
import uuid
from typing import Any

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

from app.core.permissions import check_permission
from app.plugins.builtins.kommunikation.models import CommParticipant

logger = logging.getLogger(__name__)


class CommRBAC:
    """Chat-internal RBAC, integrated with the existing permission system."""

    @staticmethod
    async def get_user_role(
        db: AsyncSession,
        conversation_id: uuid.UUID,
        user_id: uuid.UUID,
    ) -> str | None:
        """Get the user's role in a conversation, or None if not a participant."""
        result = await db.execute(
            select(CommParticipant).where(
                CommParticipant.conversation_id == conversation_id,
                CommParticipant.participant_id == user_id,
                CommParticipant.participant_type == "user",
                CommParticipant.left_at.is_(None),
            )
        )
        participant = result.scalar_one_or_none()
        return participant.role if participant else None

    @staticmethod
    async def is_participant(
        db: AsyncSession,
        conversation_id: uuid.UUID,
        user_id: uuid.UUID,
    ) -> bool:
        """Check if a user is an active participant in a conversation."""
        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
        return role is not None

    @staticmethod
    async def can_user_write(
        db: AsyncSession,
        conversation_id: uuid.UUID,
        current_user: dict[str, Any],
    ) -> bool:
        """Check if user can write messages.

        1. System permission 'comm:write' via check_permission
        2. is_system_admin → always allowed
        3. User must be participant with role 'admin' or 'member'
        """
        if current_user.get("is_system_admin"):
            return True
        if not check_permission(current_user, "comm:write"):
            return False
        user_id = uuid.UUID(current_user["user_id"])
        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
        return role in ("admin", "member")

    @staticmethod
    async def can_user_manage(
        db: AsyncSession,
        conversation_id: uuid.UUID,
        current_user: dict[str, Any],
    ) -> bool:
        """Check if user can manage participants.

        1. System permission 'comm:manage' via check_permission
        2. is_system_admin → always allowed
        3. User must be conversation admin
        """
        if current_user.get("is_system_admin"):
            return True
        if not check_permission(current_user, "comm:manage"):
            return False
        user_id = uuid.UUID(current_user["user_id"])
        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
        return role == "admin"

    @staticmethod
    async def can_user_delete(
        db: AsyncSession,
        conversation_id: uuid.UUID,
        current_user: dict[str, Any],
        is_own_message: bool = False,
    ) -> bool:
        """Check if user can delete messages or conversation.

        1. is_system_admin → always allowed
        2. For own messages: comm:write suffices
        3. For others' messages: comm:delete + conversation admin
        4. For conversation: comm:delete + admin role
        """
        if current_user.get("is_system_admin"):
            return True
        if is_own_message:
            return check_permission(current_user, "comm:write")
        if not check_permission(current_user, "comm:delete"):
            return False
        user_id = uuid.UUID(current_user["user_id"])
        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
        return role == "admin"

    @staticmethod
    async def can_user_create(current_user: dict[str, Any]) -> bool:
        """Check if user can create conversations."""
        if current_user.get("is_system_admin"):
            return True
        return check_permission(current_user, "comm:create")

    @staticmethod
    async def can_user_change_role(
        db: AsyncSession,
        conversation_id: uuid.UUID,
        current_user: dict[str, Any],
    ) -> bool:
        """Check if user can change participant roles.

        Requires comm:admin permission + conversation admin role.
        """
        if current_user.get("is_system_admin"):
            return True
        if not check_permission(current_user, "comm:admin"):
            return False
        user_id = uuid.UUID(current_user["user_id"])
        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
        return role == "admin"