gogo-telefon/gogo/auth.py

"""Session-based auth (§4): owner accounts (one per salon) + super-admins.

No public registration — accounts are created by the super-admin. Sessions are
signed cookies (itsdangerous). Admins can impersonate a tenant ("open dashboard
as salon") for support/onboarding (§11).
"""

from __future__ import annotations

import uuid
from typing import Any

import bcrypt
from fastapi import Depends, HTTPException, Request, Response
from itsdangerous import BadSignature, URLSafeTimedSerializer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from gogo.config import get_settings
from gogo.db import get_session
from gogo.models import Admin, Tenant, User


def hash_password(plain: str) -> str:
    return bcrypt.hashpw(plain.encode(), bcrypt.gensalt()).decode()


def verify_password(plain: str, hashed: str) -> bool:
    try:
        return bcrypt.checkpw(plain.encode(), hashed.encode())
    except ValueError:
        return False


def _serializer() -> URLSafeTimedSerializer:
    return URLSafeTimedSerializer(get_settings().secret_key, salt="session")


def set_session(response: Response, data: dict[str, Any]) -> None:
    s = get_settings()
    response.set_cookie(
        s.session_cookie,
        _serializer().dumps(data),
        max_age=s.session_max_age,
        httponly=True,
        samesite="lax",
    )


def clear_session(response: Response) -> None:
    response.delete_cookie(get_settings().session_cookie)


def read_session(request: Request) -> dict[str, Any] | None:
    raw = request.cookies.get(get_settings().session_cookie)
    if not raw:
        return None
    try:
        return _serializer().loads(raw, max_age=get_settings().session_max_age)
    except BadSignature:
        return None


class OwnerContext:
    def __init__(self, user: User | None, tenant: Tenant, impersonated_by: str | None = None):
        self.user = user
        self.tenant = tenant
        self.impersonated_by = impersonated_by  # admin email when impersonating


async def current_owner(
    request: Request, session: AsyncSession = Depends(get_session)
) -> OwnerContext:
    """Dependency: require an owner session (or an admin impersonating one)."""
    data = read_session(request)
    if data and data.get("kind") == "owner":
        user = (
            await session.execute(select(User).where(User.id == uuid.UUID(data["user_id"])))
        ).scalar_one_or_none()
        if user and user.active:
            tenant = (
                await session.execute(select(Tenant).where(Tenant.id == user.tenant_id))
            ).scalar_one()
            return OwnerContext(user, tenant)
    if data and data.get("kind") == "admin" and data.get("impersonate_tenant"):
        tenant = (
            await session.execute(
                select(Tenant).where(Tenant.id == uuid.UUID(data["impersonate_tenant"]))
            )
        ).scalar_one_or_none()
        if tenant:
            return OwnerContext(None, tenant, impersonated_by=data.get("email"))
    raise HTTPException(status_code=303, headers={"Location": "/login"})


async def current_admin(
    request: Request, session: AsyncSession = Depends(get_session)
) -> Admin:
    data = read_session(request)
    if data and data.get("kind") == "admin":
        admin = (
            await session.execute(select(Admin).where(Admin.id == uuid.UUID(data["admin_id"])))
        ).scalar_one_or_none()
        if admin and admin.active:
            return admin
    raise HTTPException(status_code=303, headers={"Location": "/admin/login"})