leocrm/app/deps.py

"""FastAPI dependencies: auth, db, tenant context, RBAC."""

from __future__ import annotations

import logging
import uuid
from typing import Any

import redis.asyncio as aioredis
from fastapi import Depends, HTTPException, Request, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import get_settings
from app.core.auth import get_redis, get_session_data, refresh_session_ttl
from app.core.db import get_db, set_tenant_context

logger = logging.getLogger(__name__)

# Known write-permission modules — used by require_write() to check
# specific permissions instead of broad wildcards like *:write
_WRITE_PERMISSIONS = [
    "contacts:write",
    "contacts:create",
    "users:write",
    "roles:write",
    "audit:write",
    "attachments:write",
    "workflows:write",
    "sequences:write",
    "addresses:write",
    "taxes:write",
    "currencies:write",
    "notifications:write",
    "import_export:write",
    "user_preferences:write",
]


async def get_redis_dep() -> aioredis.Redis:
    """FastAPI dependency for Redis client."""
    return get_redis()


async def get_current_user(
    request: Request,
    db: AsyncSession = Depends(get_db),
    redis: aioredis.Redis = Depends(get_redis_dep),
) -> dict[str, Any]:
    """Get the current authenticated user from session cookie.

    Returns session data dict with user_id, tenant_id, email, name, role,
    and resolved permissions from Redis cache.
    """
    settings = get_settings()

    session_id = request.cookies.get(settings.session_cookie_name)
    if not session_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail={"detail": "Not authenticated", "code": "not_authenticated"},
        )

    session_data = await get_session_data(redis, session_id)
    if session_data is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail={"detail": "Session expired or invalid", "code": "session_invalid"},
        )

    # Sliding session: extend TTL on each authenticated request
    await refresh_session_ttl(redis, session_id)

    if not session_data.get("is_active", True):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail={"detail": "User account deactivated", "code": "user_inactive"},
        )

    # Set RLS tenant context
    tenant_id = uuid.UUID(session_data["tenant_id"])
    await set_tenant_context(db, tenant_id)

    # Load resolved permissions from cache (or DB on miss)
    from app.core.permissions import get_cached_permissions

    user_id = uuid.UUID(session_data["user_id"])
    resolved = await get_cached_permissions(db, redis, user_id, tenant_id)
    session_data["permissions"] = resolved.get("permissions", [])
    session_data["denied_permissions"] = resolved.get("denied", [])
    session_data["field_permissions"] = resolved.get("field_permissions", {})
    session_data["is_system_admin"] = resolved.get("is_system_admin", False)

    return session_data


async def require_admin(
    current_user: dict[str, Any] = Depends(get_current_user),
) -> dict[str, Any]:
    """Require admin role (legacy + new permission system).

    Legacy role string 'admin' is deprecated — log a warning when used.
    New system uses is_system_admin or *:* permission.
    """
    if current_user.get("is_system_admin"):
        return current_user

    # Legacy role string fallback — deprecated
    if current_user.get("role") == "admin":
        logger.warning(
            "Legacy role string 'admin' used for user=%s — deprecated, "
            "migrate to is_system_admin or *:* permission",
            current_user.get("user_id"),
        )
        return current_user

    # New permission system check
    from app.core.permissions import check_permission

    if check_permission(current_user, "*:*"):
        return current_user

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail={"detail": "Admin access required", "code": "forbidden"},
    )


async def require_write(
    current_user: dict[str, Any] = Depends(get_current_user),
) -> dict[str, Any]:
    """Require write permission (admin, editor, or custom role with write perms).

    Legacy role strings 'admin'/'editor' are deprecated — log a warning when used.
    New system checks specific module:write permissions instead of broad wildcards.
    """
    if current_user.get("is_system_admin"):
        return current_user

    # Legacy role string fallback — deprecated
    role = current_user.get("role", "viewer")
    if role in ("admin", "editor"):
        logger.warning(
            "Legacy role string '%s' used for user=%s in require_write — deprecated, "
            "migrate to specific module:write permissions",
            role, current_user.get("user_id"),
        )
        return current_user

    # Check via permission system for specific write permissions
    from app.core.permissions import check_permission

    for perm in _WRITE_PERMISSIONS:
        if check_permission(current_user, perm):
            return current_user

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail={"detail": "Write access required", "code": "forbidden"},
    )


def require_permission(permission: str):
    """FastAPI dependency factory: require a specific permission.

    Usage:
        @router.get("/contacts", dependencies=[Depends(require_permission("contacts:read"))])
    """
    async def _check(
        current_user: dict[str, Any] = Depends(get_current_user),
    ) -> dict[str, Any]:
        if current_user.get("is_system_admin"):
            return current_user
        from app.core.permissions import check_permission

        if check_permission(current_user, permission):
            return current_user
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail={
                "detail": f"Permission '{permission}' required",
                "code": "forbidden",
            },
        )

    return _check


def require_field_access(module: str, field: str, default: str = "read"):
    """FastAPI dependency factory: require field-level access.

    Usage:
        @router.get("/contacts/{id}", dependencies=[Depends(require_field_access("contacts", "annual_revenue"))])
    """
    async def _check(
        current_user: dict[str, Any] = Depends(get_current_user),
    ) -> dict[str, Any]:
        if current_user.get("is_system_admin"):
            return current_user
        from app.core.permissions import check_field_access

        access = check_field_access(current_user, module, field, default)
        if access == "hidden":
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail={
                    "detail": f"Field '{field}' is hidden",
                    "code": "field_hidden",
                },
            )
        return current_user

    return _check


async def get_tenant_id(
    current_user: dict[str, Any] = Depends(get_current_user),
) -> uuid.UUID:
    """Extract tenant_id from current user session."""
    return uuid.UUID(current_user["tenant_id"])


async def get_current_user_id(
    current_user: dict[str, Any] = Depends(get_current_user),
) -> uuid.UUID:
    """Extract user_id from current user session."""
    return uuid.UUID(current_user["user_id"])