leocrm/app/core/api_token.py

"""API Token Service — create, verify, revoke, list Bearer tokens.

Uses ApiToken model with token_hash (SHA-256). Tokens are shown once at creation
and never stored in plaintext. Verification hashes the incoming token and
matches against the database.
"""

from __future__ import annotations

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

from sqlalchemy import select, update, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.auth import ApiToken
from app.models.user import User, UserTenant


def _hash_token(token: str) -> str:
    """Hash a plaintext token with SHA-256."""
    return hashlib.sha256(token.encode()).hexdigest()


def _generate_token() -> str:
    """Generate a secure random token (URL-safe, 32 bytes)."""
    return secrets.token_urlsafe(32)


async def create_api_token(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    name: str,
    scopes: list[str] | None = None,
    expires_at: datetime | None = None,
) -> dict[str, Any]:
    """Create a new API token. Returns the plaintext token ONCE."""
    plaintext = _generate_token()
    token_hash = _hash_token(plaintext)

    token = ApiToken(
        tenant_id=tenant_id,
        user_id=user_id,
        token_hash=token_hash,
        name=name,
        scopes=scopes or [],
        expires_at=expires_at,
    )
    db.add(token)
    await db.flush()
    await db.refresh(token)

    return {
        "id": str(token.id),
        "token": plaintext,  # Only returned once at creation
        "name": token.name,
        "scopes": token.scopes,
        "expires_at": token.expires_at.isoformat() if token.expires_at else None,
        "created_at": token.created_at.isoformat() if token.created_at else None,
    }


async def verify_api_token(
    db: AsyncSession, token: str
) -> dict[str, Any] | None:
    """Verify a Bearer token. Returns user context dict or None.

    Checks:
    - Token hash matches a database record
    - Token is not revoked (revoked_at is NULL)
    - Token is not expired (expires_at is NULL or in the future)
    - User is active
    - User has an active membership in the token's tenant
    """
    token_hash = _hash_token(token)

    q = select(ApiToken).where(
        ApiToken.token_hash == token_hash,
        ApiToken.revoked_at.is_(None),
    )
    result = await db.execute(q)
    api_token = result.scalar_one_or_none()

    if api_token is None:
        return None

    # Check expiry
    now = datetime.now(UTC)
    if api_token.expires_at is not None:
        expires_at = api_token.expires_at
        if expires_at.tzinfo is None:
            expires_at = expires_at.replace(tzinfo=UTC)
        if now > expires_at:
            return None

    # Load user
    user_q = select(User).where(User.id == api_token.user_id, User.is_active == True)  # noqa: E712
    user_result = await db.execute(user_q)
    user = user_result.scalar_one_or_none()
    if user is None:
        return None

    # Check active membership
    ut_q = select(UserTenant).where(
        UserTenant.user_id == user.id,
        UserTenant.tenant_id == api_token.tenant_id,
        UserTenant.status == "active",
    )
    ut_result = await db.execute(ut_q)
    ut = ut_result.scalar_one_or_none()
    if ut is None:
        return None

    # Update last_used_at (non-blocking)
    await db.execute(
        update(ApiToken)
        .where(ApiToken.id == api_token.id)
        .values(last_used_at=now)
    )
    await db.flush()

    # Build user context dict (same shape as get_current_user)
    return {
        "user_id": str(user.id),
        "tenant_id": str(api_token.tenant_id),
        "email": user.email,
        "name": user.name,
        "role": ut.role,
        "is_system_admin": user.is_system_admin,
        "permissions": [],  # Loaded by require_permission if needed
        "_auth_method": "api_token",
        "_token_id": str(api_token.id),
        "_token_scopes": api_token.scopes or [],
    }


async def revoke_api_token(
    db: AsyncSession, tenant_id: uuid.UUID, token_id: uuid.UUID
) -> bool:
    """Revoke an API token."""
    now = datetime.now(UTC)
    result = await db.execute(
        update(ApiToken)
        .where(
            ApiToken.id == token_id,
            ApiToken.tenant_id == tenant_id,
            ApiToken.revoked_at.is_(None),
        )
        .values(revoked_at=now)
    )
    await db.flush()
    return result.rowcount > 0


async def list_api_tokens(
    db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID
) -> list[dict[str, Any]]:
    """List all API tokens for a user (without token hashes)."""
    q = select(ApiToken).where(
        ApiToken.tenant_id == tenant_id,
        ApiToken.user_id == user_id,
        ApiToken.revoked_at.is_(None),
    ).order_by(ApiToken.created_at.desc())
    result = await db.execute(q)
    tokens = result.scalars().all()
    return [
        {
            "id": str(t.id),
            "name": t.name,
            "scopes": t.scopes or [],
            "expires_at": t.expires_at.isoformat() if t.expires_at else None,
            "last_used_at": t.last_used_at.isoformat() if t.last_used_at else None,
            "created_at": t.created_at.isoformat() if t.created_at else None,
        }
        for t in tokens
    ]