leocrm/app/services/permission_resolver.py

"""Permission resolution logic — effective access, visibility, batch resolution.

Extracted from entity_permission_service.py for modularity.
"""

from __future__ import annotations

import uuid
from datetime import datetime, UTC

from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.permissions import PERM_RANK as _PERM_RANK
from app.models.entity_permission import EntityPermission
from app.models.group import UserGroup
from app.models.user import User, UserTenant


def _rank(level: str) -> int:
    return _PERM_RANK.get(level, 0)


def _get_entity_model(entity_type: str) -> type:
    """Get SQLAlchemy model class for entity_type, or raise ValueError.

    Uses lazy import to avoid circular dependency with entity_permission_service.
    """
    from app.services.entity_permission_service import ENTITY_MODELS
    model = ENTITY_MODELS.get(entity_type)
    if model is None:
        raise ValueError(f"Unknown entity type: {entity_type}")
    return model


async def get_effective_access(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    entity_type: str,
    entity_id: uuid.UUID,
) -> str:
    """Get the effective access level for a user on a specific entity.

    Resolution (highest wins):
    1. System admin → 'delete' (full access)
    2. Owner → 'owner' (from owner_id on the entity table)
    3. Direct user permission
    4. Group permission (via user_groups)
    5. Role permission (via user_tenants.role_id)
    6. Guest permission (principal_type='guest')
    7. owner_id IS NULL → 'read' (tenant-owned, visible to all with module permission)
    8. No access → 'none'

    Returns: 'none' | 'read' | 'write' | 'admin' | 'delete' | 'owner'
    """
    # Check system admin
    user_q = await db.execute(
        select(User.is_system_admin).where(User.id == user_id)
    )
    if user_q.scalar():
        return "delete"

    # Check ownership — load the entity's owner_id via SQLAlchemy model (safe from SQL injection)
    model = _get_entity_model(entity_type)
    owner_q = await db.execute(
        select(model.owner_id).where(model.id == entity_id).where(model.tenant_id == tenant_id)
    )
    owner_row = owner_q.first()
    if not owner_row:
        return "none"

    owner_id = owner_row[0]
    if owner_id == user_id:
        return "owner"

    # Get user's groups and role
    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()

    # Build principal conditions
    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,
            )
        )
    # Guest permission — guests are now regular users with role=guest
    # (Guest users may have no groups/roles, only direct entity_permissions)
    principal_conditions.append(
        and_(
            EntityPermission.principal_type == "guest",
            EntityPermission.principal_id == user_id,
        )
    )

    # Query permissions
    now = datetime.now(UTC)
    perm_q = await db.execute(
        select(EntityPermission.permission_level)
        .where(EntityPermission.entity_type == entity_type)
        .where(EntityPermission.entity_id == entity_id)
        .where(EntityPermission.tenant_id == tenant_id)
        .where(or_(*principal_conditions))
        .where(
            or_(
                EntityPermission.expires_at.is_(None),
                EntityPermission.expires_at > now,
            )
        )
    )

    best_level = "none"
    for (level,) in perm_q:
        if _rank(level) > _rank(best_level):
            best_level = level

    # If owner_id is NULL (tenant-owned), user with module permission gets at least 'read'
    if best_level == "none" and owner_id is None:
        return "read"

    return best_level


async def get_visible_ids(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    entity_type: str,
) -> tuple[set[uuid.UUID], dict[uuid.UUID, str]]:
    """Get all visible entity IDs for a user and their access levels.

    Returns (visible_ids, access_map) where access_map is
    entity_id → access_level string.

    Resolution:
    1. System admin → all entities at 'delete' level
    2. Owned entities → 'owner'
    3. Entities with direct/group/role permissions → permission level
    4. Tenant-owned entities (owner_id IS NULL) → 'read'
    """
    # Check system admin
    user_q = await db.execute(
        select(User.is_system_admin).where(User.id == user_id)
    )
    if user_q.scalar():
        # Return all entity IDs
        model = _get_entity_model(entity_type)
        admin_q = select(model.id).where(model.tenant_id == tenant_id)
        if hasattr(model, 'deleted_at'):
            admin_q = admin_q.where(model.deleted_at.is_(None))
        all_q = await db.execute(admin_q)
        all_ids = {row[0] for row in all_q}
        return all_ids, {eid: "delete" for eid in all_ids}

    # Get user's groups and role
    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()

    # 1. Owned entities
    model = _get_entity_model(entity_type)
    owned_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id == user_id)
    if hasattr(model, 'deleted_at'):
        owned_q_builder = owned_q_builder.where(model.deleted_at.is_(None))
    owned_q = await db.execute(owned_q_builder)
    visible: set[uuid.UUID] = set()
    access_map: dict[uuid.UUID, str] = {}
    for (eid,) in owned_q:
        visible.add(eid)
        access_map[eid] = "owner"

    # 2. Tenant-owned entities (owner_id IS NULL)
    tenant_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id.is_(None))
    if hasattr(model, 'deleted_at'):
        tenant_q_builder = tenant_q_builder.where(model.deleted_at.is_(None))
    tenant_owned_q = await db.execute(tenant_q_builder)
    for (eid,) in tenant_owned_q:
        if eid not in visible:
            visible.add(eid)
            access_map[eid] = "read"

    # 3. Permission-based access
    now = datetime.now(UTC)
    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,
            )
        )
    # Guest permission — guest users have no groups/roles
    principal_conditions.append(
        and_(
            EntityPermission.principal_type == "guest",
            EntityPermission.principal_id == user_id,
        )
    )

    perm_q = await db.execute(
        select(EntityPermission.entity_id, EntityPermission.permission_level)
        .where(EntityPermission.entity_type == entity_type)
        .where(EntityPermission.tenant_id == tenant_id)
        .where(or_(*principal_conditions))
        .where(
            or_(
                EntityPermission.expires_at.is_(None),
                EntityPermission.expires_at > now,
            )
        )
    )
    for eid, level in perm_q:
        if eid not in visible or _rank(level) > _rank(access_map.get(eid, "none")):
            visible.add(eid)
            access_map[eid] = level

    return visible, access_map


async def batch_get_effective_access(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    entity_type: str,
    entity_ids: list[uuid.UUID],
) -> dict[uuid.UUID, str]:
    """Batch resolution: get access levels for multiple entities at once.

    Much more efficient than calling get_effective_access() in a loop.
    """
    if not entity_ids:
        return {}

    # Check system admin
    user_q = await db.execute(
        select(User.is_system_admin).where(User.id == user_id)
    )
    if user_q.scalar():
        return {eid: "delete" for eid in entity_ids}

    # Get user's groups and role
    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()

    result: dict[uuid.UUID, str] = {}

    # 1. Check ownership via SQLAlchemy model (safe from SQL injection)
    model = _get_entity_model(entity_type)
    batch_q = select(model.id, model.owner_id).where(model.id.in_(entity_ids)).where(model.tenant_id == tenant_id)
    if hasattr(model, 'deleted_at'):
        batch_q = batch_q.where(model.deleted_at.is_(None))
    owner_q = await db.execute(batch_q)
    for eid, owner_id in owner_q:
        if owner_id == user_id:
            result[eid] = "owner"
        elif owner_id is None:
            result[eid] = "read"

    # 2. Check permissions
    now = datetime.now(UTC)
    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,
            )
        )
    # Guest permission — guest users have no groups/roles
    principal_conditions.append(
        and_(
            EntityPermission.principal_type == "guest",
            EntityPermission.principal_id == user_id,
        )
    )

    perm_q = await db.execute(
        select(EntityPermission.entity_id, EntityPermission.permission_level)
        .where(EntityPermission.entity_type == entity_type)
        .where(EntityPermission.entity_id.in_(entity_ids))
        .where(EntityPermission.tenant_id == tenant_id)
        .where(or_(*principal_conditions))
        .where(
            or_(
                EntityPermission.expires_at.is_(None),
                EntityPermission.expires_at > now,
            )
        )
    )
    for eid, level in perm_q:
        current = result.get(eid, "none")
        if _rank(level) > _rank(current):
            result[eid] = level

    # Fill in 'none' for entities not found
    for eid in entity_ids:
        if eid not in result:
            result[eid] = "none"

    return result


async def check_entity_access(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    entity_type: str,
    entity_id: uuid.UUID,
    required_level: str = "read",
) -> bool:
    """Check if user has at least the required access level on an entity."""
    access = await get_effective_access(db, tenant_id, user_id, entity_type, entity_id)
    return _rank(access) >= _rank(required_level)