"""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"})