leocrm/app/core/auth.py

"""Session-based authentication, password hashing, and RBAC."""

from __future__ import annotations

import hashlib
import secrets
import uuid
from datetime import UTC, datetime, timedelta
from typing import Any

import redis.asyncio as aioredis
from passlib.context import CryptContext
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import get_settings
from app.models.session import Session as SessionModel
from app.models.user import User

_pwd_context = CryptContext(
    schemes=["bcrypt"], deprecated="auto", bcrypt__rounds=get_settings().bcrypt_rounds
)


def hash_password(password: str) -> str:
    """Hash a password using bcrypt."""
    return _pwd_context.hash(password)


def verify_password(password: str, password_hash: str) -> bool:
    """Verify a password against a bcrypt hash."""
    return _pwd_context.verify(password, password_hash)


def generate_session_token() -> str:
    """Generate a cryptographically secure session token."""
    return secrets.token_urlsafe(32)


def generate_csrf_token() -> str:
    """Generate a CSRF token."""
    return secrets.token_urlsafe(32)


def hash_token(token: str) -> str:
    """SHA-256 hash a token for storage."""
    return hashlib.sha256(token.encode()).hexdigest()


def get_redis() -> aioredis.Redis:
    """Get a Redis client instance."""
    return aioredis.from_url(get_settings().redis_url, decode_responses=True)


async def create_session(
    db: AsyncSession,
    redis: aioredis.Redis,
    user: User,
    tenant_id: uuid.UUID,
) -> tuple[str, str]:
    """Create a session in Redis (runtime) and PostgreSQL (audit trail).
    Returns (session_id, csrf_token).
    """
    settings = get_settings()
    session_id = str(uuid.uuid4())
    csrf_token = generate_csrf_token()
    expires_at = datetime.now(UTC) + timedelta(seconds=settings.session_ttl_seconds)

    # Redis runtime session
    session_data: dict[str, Any] = {
        "user_id": str(user.id),
        "tenant_id": str(tenant_id),
        "email": user.email,
        "name": user.name,
        "role": user.role,
        "is_system_admin": user.is_system_admin,
        "csrf_token": csrf_token,
        "is_active": user.is_active,
    }
    import json

    await redis.setex(
        f"session:{session_id}",
        settings.session_ttl_seconds,
        json.dumps(session_data),
    )

    # PostgreSQL audit trail
    audit_record = SessionModel(
        id=uuid.UUID(session_id),
        user_id=user.id,
        tenant_id=tenant_id,
        csrf_token=csrf_token,
        expires_at=expires_at,
    )
    db.add(audit_record)
    await db.flush()

    return session_id, csrf_token


async def get_session_data(redis: aioredis.Redis, session_id: str) -> dict[str, Any] | None:
    """Retrieve session data from Redis."""
    import json

    raw = await redis.get(f"session:{session_id}")
    if raw is None:
        return None
    return json.loads(raw)


async def invalidate_session(redis: aioredis.Redis, session_id: str) -> None:
    """Delete a session from Redis (logout). PostgreSQL record persists."""
    await redis.delete(f"session:{session_id}")


async def update_session_tenant(
    redis: aioredis.Redis,
    session_id: str,
    new_tenant_id: uuid.UUID,
) -> dict[str, Any] | None:
    """Update the active tenant in a Redis session."""
    import json

    settings = get_settings()
    raw = await redis.get(f"session:{session_id}")
    if raw is None:
        return None
    data = json.loads(raw)
    data["tenant_id"] = str(new_tenant_id)
    ttl = await redis.ttl(f"session:{session_id}")
    if ttl <= 0:
        ttl = settings.session_ttl_seconds
    await redis.setex(f"session:{session_id}", ttl, json.dumps(data))
    return data


def check_permission(
    role_name: str, module: str, action: str, permissions: dict | None = None
) -> bool:
    """Check if a role has permission for a module+action.
    Built-in roles: admin (all), editor (read+write), viewer (read only).
    Custom roles use the permissions dict.
    """
    if role_name == "admin":
        return True
    if role_name == "editor":
        if action in ("read", "write", "create", "update"):
            return True
        return False
    if role_name == "viewer":
        return action == "read"
    # Custom role — check permissions dict
    if permissions:
        module_perms = permissions.get(module, {})
        return bool(module_perms.get(action, False))
    return False


def filter_fields_by_permission(
    data: dict[str, Any],
    field_permissions: dict[str, str],
    role_name: str,
) -> dict[str, Any]:
    """Filter response fields based on field-level permissions.
    field_permissions: {"annual_revenue": "hidden"} → removed for non-admin.
    """
    if role_name == "admin":
        return data
    result = {}
    for key, value in data.items():
        perm = field_permissions.get(key)
        if perm == "hidden":
            continue
        result[key] = value
    return result