libra/auth.py

"""
Centralized Authorization Module for Libra Extension.

Provides consistent, secure authorization patterns across all endpoints.

Key concepts:
- AuthContext: Captures all authorization state for a request
- Dependencies: FastAPI dependencies for endpoint protection
- Permission checks: Consistent resource-level access control

Usage:
    from .auth import require_super_user, require_authenticated, AuthContext

    @router.get("/api/v1/admin-endpoint")
    async def admin_endpoint(auth: AuthContext = Depends(require_super_user)):
        # Only super users can access
        pass

    @router.get("/api/v1/user-data")
    async def user_data(auth: AuthContext = Depends(require_authenticated)):
        # Any authenticated user
        user_id = auth.user_id
        pass
"""

from dataclasses import dataclass
from functools import wraps
from http import HTTPStatus
from typing import Optional

from fastapi import Depends, HTTPException
from lnbits.core.models import WalletTypeInfo
from lnbits.decorators import require_admin_key, require_invoice_key
from lnbits.settings import settings as lnbits_settings
from loguru import logger

from .crud import get_account, get_user_permissions
from .models import PermissionType


@dataclass
class AuthContext:
    """
    Authorization context for a request.

    Contains all information needed to make authorization decisions.
    Use this instead of directly accessing wallet/user properties scattered
    throughout endpoint code.
    """
    user_id: str
    wallet_id: str
    is_super_user: bool
    wallet: WalletTypeInfo

    @property
    def is_admin(self) -> bool:
        """
        Check if user is a Libra admin (super user).

        Note: In Libra, admin = super_user. There's no separate admin concept.
        """
        return self.is_super_user

    def require_super_user(self) -> None:
        """Raise HTTPException if not super user."""
        if not self.is_super_user:
            raise HTTPException(
                status_code=HTTPStatus.FORBIDDEN,
                detail="Super user access required"
            )

    def require_self_or_super_user(self, target_user_id: str) -> None:
        """
        Require that user is accessing their own data or is super user.

        Args:
            target_user_id: The user ID being accessed

        Raises:
            HTTPException: If user is neither the target nor super user
        """
        if not self.is_super_user and self.user_id != target_user_id:
            raise HTTPException(
                status_code=HTTPStatus.FORBIDDEN,
                detail="Access denied: you can only access your own data"
            )


def _build_auth_context(wallet: WalletTypeInfo) -> AuthContext:
    """Build AuthContext from wallet info."""
    user_id = wallet.wallet.user
    return AuthContext(
        user_id=user_id,
        wallet_id=wallet.wallet.id,
        is_super_user=user_id == lnbits_settings.super_user,
        wallet=wallet,
    )


# ===== FastAPI Dependencies =====

async def require_authenticated(
    wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> AuthContext:
    """
    Require authentication (invoice key minimum).

    Returns AuthContext with user information.
    Use for read-only access to user's own data.
    """
    return _build_auth_context(wallet)


async def require_authenticated_write(
    wallet: WalletTypeInfo = Depends(require_admin_key),
) -> AuthContext:
    """
    Require authentication with write permissions (admin key).

    Returns AuthContext with user information.
    Use for write operations on user's own data.
    """
    return _build_auth_context(wallet)


async def require_super_user(
    wallet: WalletTypeInfo = Depends(require_admin_key),
) -> AuthContext:
    """
    Require super user access.

    Raises HTTPException 403 if not super user.
    Use for Libra admin operations.
    """
    auth = _build_auth_context(wallet)
    if not auth.is_super_user:
        logger.warning(
            f"Super user access denied for user {auth.user_id[:8]} "
            f"attempting admin operation"
        )
        raise HTTPException(
            status_code=HTTPStatus.FORBIDDEN,
            detail="Super user access required"
        )
    return auth


# ===== Resource Access Checks =====

async def can_access_account(
    auth: AuthContext,
    account_id: str,
    permission_type: PermissionType = PermissionType.READ,
) -> bool:
    """
    Check if user can access an account.

    Access is granted if:
    1. User is super user (full access)
    2. User owns the account (user-specific accounts like Assets:Receivable:User-abc123)
    3. User has explicit permission for the account

    Args:
        auth: The authorization context
        account_id: The account ID to check
        permission_type: The type of access needed (READ, SUBMIT_EXPENSE, MANAGE)

    Returns:
        True if access is allowed, False otherwise
    """
    # Super users have full access
    if auth.is_super_user:
        return True

    # Check if this is the user's own account. Match the User-{short}
    # segment exactly — a substring test also matched account names that
    # merely CONTAIN it (e.g. "Expenses:Misc-User-deadbeef"), granting
    # access to unrelated accounts.
    account = await get_account(account_id)
    if account:
        user_segment = f"User-{auth.user_id[:8]}"
        if user_segment in account.name.split(":"):
            return True

    # Check explicit permissions
    permissions = await get_user_permissions(auth.user_id)
    for perm in permissions:
        if perm.account_id == account_id:
            # Check if permission type is sufficient
            if perm.permission_type == PermissionType.MANAGE:
                return True  # MANAGE grants all access
            if perm.permission_type == permission_type:
                return True
            if (
                permission_type == PermissionType.READ
                and perm.permission_type in [PermissionType.SUBMIT_EXPENSE, PermissionType.MANAGE]
            ):
                return True  # Higher permissions include READ

    return False


async def require_account_access(
    auth: AuthContext,
    account_id: str,
    permission_type: PermissionType = PermissionType.READ,
) -> None:
    """
    Require access to an account, raising HTTPException if denied.

    Args:
        auth: The authorization context
        account_id: The account ID to check
        permission_type: The type of access needed

    Raises:
        HTTPException: If access is denied
    """
    if not await can_access_account(auth, account_id, permission_type):
        logger.warning(
            f"Account access denied: user {auth.user_id[:8]} "
            f"attempted {permission_type.value} on account {account_id}"
        )
        raise HTTPException(
            status_code=HTTPStatus.FORBIDDEN,
            detail=f"Access denied to account {account_id}"
        )


async def can_access_user_data(auth: AuthContext, target_user_id: str) -> bool:
    """
    Check if user can access another user's data.

    Access is granted if:
    1. User is super user
    2. User is accessing their own data

    Args:
        auth: The authorization context
        target_user_id: The user ID whose data is being accessed

    Returns:
        True if access is allowed
    """
    if auth.is_super_user:
        return True

    # Users can access their own data. Full-ID equality ONLY: an 8-char
    # prefix comparison is a 32-bit space, and any prefix collision (or a
    # deliberately crafted short target id) let one user read another's
    # data. Callers must pass full user ids.
    if auth.user_id == target_user_id:
        return True

    return False


async def require_user_data_access(
    auth: AuthContext,
    target_user_id: str,
) -> None:
    """
    Require access to a user's data, raising HTTPException if denied.

    Args:
        auth: The authorization context
        target_user_id: The user ID whose data is being accessed

    Raises:
        HTTPException: If access is denied
    """
    if not await can_access_user_data(auth, target_user_id):
        logger.warning(
            f"User data access denied: user {auth.user_id[:8]} "
            f"attempted to access data for user {target_user_id[:8]}"
        )
        raise HTTPException(
            status_code=HTTPStatus.FORBIDDEN,
            detail="Access denied: you can only access your own data"
        )


# ===== Utility Functions =====

def get_user_id_from_wallet(wallet: WalletTypeInfo) -> str:
    """
    Get user ID from wallet info.

    IMPORTANT: Always use wallet.wallet.user (not wallet.wallet.id).
    - wallet.wallet.user = the user's ID
    - wallet.wallet.id = the wallet's ID (NOT the same!)

    Args:
        wallet: The wallet type info from LNbits

    Returns:
        The user ID
    """
    return wallet.wallet.user


def is_super_user(user_id: str) -> bool:
    """
    Check if a user ID is the super user.

    Args:
        user_id: The user ID to check

    Returns:
        True if this is the super user
    """
    return user_id == lnbits_settings.super_user