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: contacts:*, *: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, func
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"

# Severity ordering for field permissions: highest wins
_FIELD_PERM_SEVERITY = {"hidden": 3, "readonly": 2, "read": 1}


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

    Supports wildcards:
    - contacts:read → exact match
    - contacts:* → all actions for contacts
    - *:read → all modules, read action
    - *:* → everything (superadmin)
    """
    if granted == required:
        return True
    g_parts = granted.split(":")
    r_parts = required.split(":")
    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]: ["contacts:read", "contacts:write"]
    - dict[str, bool]: {"contacts:read": true, "contacts:write": false}
    - dict[str, dict]: {"contacts": {"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


def _merge_field_permissions(
    existing: dict[str, dict[str, str]],
    incoming: dict[str, Any],
) -> None:
    """Merge incoming field permissions into existing dict.

    Uses 'strictest right wins': hidden > readonly > read.
    When a field already exists, the more restrictive (higher severity) value wins.
    """
    for module, fields in incoming.items():
        if not isinstance(fields, dict):
            continue
        if module not in existing:
            existing[module] = {}
        for field, perm in fields.items():
            if not isinstance(perm, str):
                continue
            perm_lower = perm.lower()
            if perm_lower not in _FIELD_PERM_SEVERITY:
                # Unknown permission level — skip with warning
                logger.warning(
                    "Unknown field permission level '%s' for %s.%s — skipping",
                    perm, module, field,
                )
                continue
            current = existing[module].get(field)
            if current is None:
                existing[module][field] = perm_lower
            else:
                # Strictest (highest severity) wins
                if _FIELD_PERM_SEVERITY[perm_lower] > _FIELD_PERM_SEVERITY.get(current, 0):
                    existing[module][field] = perm_lower


async def _get_current_permission_version(
    db: AsyncSession,
    user_id: uuid.UUID,
    tenant_id: uuid.UUID,
) -> int:
    """Get the current max permission_version from DB for cache validation.

    Uses a SAVEPOINT so that a failure here does not abort the outer transaction.
    """
    async with db.begin_nested():
        # Check role version
        ut_q = select(UserTenant.role_id).where(
            UserTenant.user_id == user_id,
            UserTenant.tenant_id == tenant_id,
        )
        ut_result = await db.execute(ut_q)
        role_id = ut_result.scalar_one_or_none()

        max_version = 0
        if role_id is not None:
            role_q = select(Role.permission_version).where(Role.id == role_id)
            role_result = await db.execute(role_q)
            role_ver = role_result.scalar_one_or_none()
            if role_ver is not None:
                max_version = max(max_version, role_ver)

        # Check group versions
        ug_q = select(UserGroup.group_id).where(
            UserGroup.user_id == user_id,
            UserGroup.tenant_id == tenant_id,
        )
        ug_result = await db.execute(ug_q)
        group_ids = [row[0] for row in ug_result.all()]

        if group_ids:
            groups_q = select(func.max(Group.permission_version)).where(
                Group.id.in_(group_ids),
                Group.deleted_at.is_(None),
            )
            groups_result = await db.execute(groups_q)
            group_max = groups_result.scalar()
            if group_max is not None:
                max_version = max(max_version, group_max)

        return max_version


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
        }
    """
    # Use SAVEPOINT for the initial query so a failure doesn't abort
    # the outer transaction.
    try:
        async with db.begin_nested():
            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
    except Exception:
        logger.warning(
            "SAVEPOINT failed for is_system_admin query (user=%s), retrying without savepoint",
            user_id,
            exc_info=True,
        )
        # Last-resort fallback: still use savepoint to isolate
        async with db.begin_nested():
            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 — use SAVEPOINT
    async with db.begin_nested():
        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:
        async with db.begin_nested():
            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 or 0)
            # Merge field permissions using strictest-wins
            if role.field_permissions:
                _merge_field_permissions(field_perms, role.field_permissions)

    # Also check built-in role string on UserTenant for backward compatibility
    if user_tenant is not None and user_tenant.role_id is None:
        legacy_role = user_tenant.role

        if legacy_role == "admin":
            allowed.add("*:*")
        elif legacy_role == "editor":
            allowed |= {
                "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",
                "user_preferences:read", "user_preferences:write",
            }
        elif legacy_role == "viewer":
            allowed |= {
                "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",
                "user_preferences:read", "user_preferences:write",
            }

    # Load group permissions
    async with db.begin_nested():
        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]
        async with db.begin_nested():
            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 or 0)
            # Merge field permissions using strictest-wins
            if group.field_permissions:
                _merge_field_permissions(field_perms, group.field_permissions)

    # 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.

    Validates the cached permission_version against the current DB version.
    If they differ, the cache entry is stale and will be re-resolved.
    """
    cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"

    raw = await redis.get(cache_key)
    if raw is not None:
        data = json.loads(raw)
        cached_version = data.get("version", -1)

        # Validate cached version against current DB version
        try:
            current_version = await _get_current_permission_version(db, user_id, tenant_id)
        except Exception:
            logger.warning(
                "Failed to query current permission_version for cache validation "
                "(user=%s, tenant=%s) — invalidating cache and re-resolving",
                user_id, tenant_id,
                exc_info=True,
            )
            # Invalidate stale cache — do NOT trust cached permissions on DB error
            await redis.delete(cache_key)
            return None  # Fall through to re-resolution from DB

        if cached_version == current_version:
            return data

        # Version mismatch — invalidate stale cache and re-resolve
        logger.info(
            "Permission cache version mismatch for user=%s tenant=%s "
            "(cached=%s, current=%s) — re-resolving",
            user_id, tenant_id, cached_version, current_version,
        )
        await redis.delete(cache_key)

    # Cache miss or stale — 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).

    Uses SCAN (non-blocking) instead of KEYS to avoid blocking Redis.
    """
    pattern = f"{CACHE_PREFIX}:*:{tenant_id}"
    batch_size = 200
    cursor: int | bytes | str = 0
    deleted_count = 0

    while True:
        cursor, keys = await redis.scan(
            cursor=cursor,
            match=pattern,
            count=batch_size,
        )
        if keys:
            await redis.delete(*keys)
            deleted_count += len(keys)
        # SCAN returns cursor as bytes or int depending on redis-py version
        cursor_int = int(cursor) if cursor else 0
        if cursor_int == 0:
            break

    logger.info(
        "Invalidated %d permission cache entries for tenant=%s",
        deleted_count, tenant_id,
    )


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 "contacts: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