leocrm/app/core/permissions.py

"""Permission resolver — resolves effective permissions for a user+tenant.

Architecture:
- Permissions are NOT stored in the session.
- Redis cache: resolved:{user_id}:{tenant_id} with 5-min TTL.
- Permission-version stamping for immediate invalidation.
- Resolution: allowed = (role ∪ groups), denied = (role.denied ∪ groups.denied), resolved = allowed − denied.
- Wildcards: companies:*, *:read, *:* (bare * is forbidden).
"""

from __future__ import annotations

import json
import logging
import uuid
from typing import Any

import redis.asyncio as aioredis
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import get_settings
from app.core.auth import get_redis
from app.models.group import Group, UserGroup
from app.models.role import Role
from app.models.user import User, UserTenant

logger = logging.getLogger(__name__)

CACHE_TTL = 300  # 5 minutes
CACHE_PREFIX = "resolved"


def _matches_permission(granted: str, required: str) -> bool:
    """Check if a granted permission matches the required permission.

    Supports wildcards:
    - companies:read → exact match
    - companies:* → all actions for companies
    - *:read → all modules, read action
    - *:* → everything (superadmin)
    """
    if granted == required:
        return True
    g_parts = granted.split(":")
    r_parts = required.split(":")
    # Wildcard * matches any single segment, but remaining segments must still match
    if len(g_parts) != len(r_parts):
        return False
    for i, g_part in enumerate(g_parts):
        if g_part == "*":
            continue
        if g_part != r_parts[i]:
            return False
    return True


def _permission_matches_any(granted_permissions: set[str], required: str) -> bool:
    """Check if any granted permission matches the required permission."""
    for granted in granted_permissions:
        if _matches_permission(granted, required):
            return True
    return False


def _normalize_permissions(permissions: Any) -> set[str]:
    """Normalize permissions from JSONB to a set of strings.

    Supports formats:
    - list[str]: ["companies:read", "contacts:write"]
    - dict[str, bool]: {"companies:read": true, "contacts:write": false}
    - dict[str, dict]: {"companies": {"read": true, "write": false}}
    """
    result: set[str] = set()
    if isinstance(permissions, list):
        for p in permissions:
            if isinstance(p, str):
                result.add(p.replace(".", ":"))
    elif isinstance(permissions, dict):
        for key, val in permissions.items():
            if isinstance(val, bool):
                if val:
                    result.add(key.replace(".", ":"))
            elif isinstance(val, dict):
                for action, enabled in val.items():
                    if enabled:
                        result.add(f"{key}:{action}".replace(".", ":"))
    return result


async def resolve_permissions(
    db: AsyncSession,
    user_id: uuid.UUID,
    tenant_id: uuid.UUID,
) -> dict[str, Any]:
    """Resolve effective permissions for a user within a tenant.

    Returns:
        {
            "permissions": set[str],       # allowed permissions
            "denied": set[str],            # explicitly denied
            "field_permissions": dict,     # {module: {field: hidden|readonly|read}}
            "is_system_admin": bool,
            "version": int,                # permission_version for cache invalidation
        }
    """
    # Check system admin first
    user_q = select(User.is_system_admin).where(User.id == user_id)
    user_result = await db.execute(user_q)
    is_system_admin = user_result.scalar() or False

    if is_system_admin:
        return {
            "permissions": {"*:*"},
            "denied": set(),
            "field_permissions": {},
            "is_system_admin": True,
            "version": 0,  # system admin doesn't need version tracking
        }

    # Load UserTenant to get role_id
    ut_q = select(UserTenant).where(
        UserTenant.user_id == user_id,
        UserTenant.tenant_id == tenant_id,
    )
    ut_result = await db.execute(ut_q)
    user_tenant = ut_result.scalar_one_or_none()

    allowed: set[str] = set()
    denied: set[str] = set()
    field_perms: dict[str, dict[str, str]] = {}
    max_version = 0

    # Load role permissions
    if user_tenant and user_tenant.role_id:
        role_q = select(Role).where(Role.id == user_tenant.role_id)
        role_result = await db.execute(role_q)
        role = role_result.scalar_one_or_none()
        if role:
            allowed |= _normalize_permissions(role.permissions)
            denied |= _normalize_permissions(role.denied_permissions)
            max_version = max(max_version, role.permission_version)
            # Merge field permissions
            if role.field_permissions:
                for module, fields in role.field_permissions.items():
                    if isinstance(fields, dict):
                        if module not in field_perms:
                            field_perms[module] = {}
                        field_perms[module].update(fields)

    # Also check legacy role string on User for backward compatibility
    if user_tenant is None or user_tenant.role_id is None:
        legacy_q = select(User.role).where(User.id == user_id)
        legacy_result = await db.execute(legacy_q)
        legacy_role = legacy_result.scalar_one_or_none()
        if legacy_role == "admin":
            allowed.add("*:*")
        elif legacy_role == "editor":
            allowed |= {"companies:read", "companies:write", "contacts:read", "contacts:write",
                        "users:read", "roles:read", "audit:read", "attachments:read",
                        "attachments:write", "workflows:read", "workflows:write",
                        "sequences:read", "sequences:write", "addresses:read", "addresses:write",
                        "taxes:read", "taxes:write", "currencies:read", "currencies:write",
                        "notifications:read", "notifications:write", "import_export:read",
                        "import_export:write"}
        elif legacy_role == "viewer":
            allowed |= {"companies:read", "contacts:read", "users:read", "roles:read",
                        "audit:read", "attachments:read", "workflows:read", "sequences:read",
                        "addresses:read", "taxes:read", "currencies:read",
                        "notifications:read", "import_export:read"}

    # Load group permissions
    ug_q = select(UserGroup).where(
        UserGroup.user_id == user_id,
        UserGroup.tenant_id == tenant_id,
    )
    ug_result = await db.execute(ug_q)
    user_groups = ug_result.scalars().all()

    if user_groups:
        group_ids = [ug.group_id for ug in user_groups]
        groups_q = select(Group).where(
            Group.id.in_(group_ids),
            Group.deleted_at.is_(None),
        )
        groups_result = await db.execute(groups_q)
        groups = groups_result.scalars().all()

        for group in groups:
            allowed |= _normalize_permissions(group.permissions)
            denied |= _normalize_permissions(group.denied_permissions)
            max_version = max(max_version, group.permission_version)
            # Merge field permissions
            if group.field_permissions:
                for module, fields in group.field_permissions.items():
                    if isinstance(fields, dict):
                        if module not in field_perms:
                            field_perms[module] = {}
                        field_perms[module].update(fields)

    # Apply deny list
    resolved = allowed - denied

    return {
        "permissions": resolved,
        "denied": denied,
        "field_permissions": field_perms,
        "is_system_admin": False,
        "version": max_version,
    }


async def get_cached_permissions(
    db: AsyncSession,
    redis: aioredis.Redis,
    user_id: uuid.UUID,
    tenant_id: uuid.UUID,
) -> dict[str, Any]:
    """Get resolved permissions from Redis cache or resolve from DB."""
    cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"

    raw = await redis.get(cache_key)
    if raw is not None:
        data = json.loads(raw)
        return data

    # Cache miss — resolve from DB
    resolved = await resolve_permissions(db, user_id, tenant_id)

    # Store in cache (convert sets to lists for JSON)
    cache_data = {
        "permissions": list(resolved["permissions"]),
        "denied": list(resolved["denied"]),
        "field_permissions": resolved["field_permissions"],
        "is_system_admin": resolved["is_system_admin"],
        "version": resolved["version"],
    }
    await redis.setex(cache_key, CACHE_TTL, json.dumps(cache_data))
    return cache_data


async def invalidate_permission_cache(
    redis: aioredis.Redis,
    user_id: uuid.UUID,
    tenant_id: uuid.UUID,
) -> None:
    """Invalidate the permission cache for a specific user+tenant."""
    cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"
    await redis.delete(cache_key)


async def invalidate_all_user_permissions(
    redis: aioredis.Redis,
    tenant_id: uuid.UUID,
) -> None:
    """Invalidate permission cache for all users in a tenant (e.g. after role/group change)."""
    pattern = f"{CACHE_PREFIX}:*:{tenant_id}"
    keys = await redis.keys(pattern)
    if keys:
        await redis.delete(*keys)


def check_permission(resolved: dict[str, Any], required: str) -> bool:
    """Check if resolved permissions grant the required permission.

    Args:
        resolved: result from get_cached_permissions or resolve_permissions
        required: permission string like "companies:read"
    """
    if resolved.get("is_system_admin"):
        return True

    permissions = set(resolved.get("permissions", []))
    denied = set(resolved.get("denied", []))

    # Check deny list first
    for d in denied:
        if _matches_permission(d, required):
            return False

    return _permission_matches_any(permissions, required)


def check_field_access(
    resolved: dict[str, Any],
    module: str,
    field: str,
    default: str = "read",
) -> str:
    """Check field-level access for a module+field.

    Returns: "hidden", "readonly", or "read"
    """
    if resolved.get("is_system_admin"):
        return "read"

    field_perms = resolved.get("field_permissions", {})
    module_perms = field_perms.get(module, {})
    return module_perms.get(field, default)


def filter_fields_by_permission(
    data: dict[str, Any],
    resolved: dict[str, Any],
    module: str,
) -> dict[str, Any]:
    """Filter response fields based on field-level permissions.

    Removes fields marked as "hidden", keeps others.
    """
    if resolved.get("is_system_admin"):
        return data

    field_perms = resolved.get("field_permissions", {})
    module_perms = field_perms.get(module, {})

    if not module_perms:
        return data

    result = {}
    for key, value in data.items():
        perm = module_perms.get(key)
        if perm == "hidden":
            continue
        result[key] = value
    return result