inkwell/src/thoughtsync/auth.py

from __future__ import annotations

import functools
import uuid

from quart import Blueprint, g, jsonify, request, session
from sqlalchemy import func, select

from .db import session_scope
from .models.user import User
from .security import hash_password, verify_password
from .settings import get_setting

bp = Blueprint("auth", __name__, url_prefix="/api/auth")

SESSION_KEY = "user_id"
MIN_PASSWORD_LEN = 8


def _serialize_user(user: User) -> dict:
    return {
        "id": str(user.id),
        "email": user.email,
        "display_name": user.display_name,
        "email_verified": user.email_verified,
        "is_admin": user.is_admin,
    }


def _session_user_id() -> uuid.UUID | None:
    raw = session.get(SESSION_KEY)
    if not raw:
        return None
    try:
        return uuid.UUID(raw)
    except (ValueError, TypeError):
        session.pop(SESSION_KEY, None)
        return None


def login_required(fn):
    """Guard: 401 unless a valid session is present. Sets g.user_id for the view."""

    @functools.wraps(fn)
    async def wrapper(*args, **kwargs):
        uid = _session_user_id()
        if uid is None:
            return jsonify({"error": "authentication required"}), 401
        g.user_id = uid
        return await fn(*args, **kwargs)

    return wrapper


def require_admin(fn):
    """Guard: 401 unauthenticated, 403 non-admin. Checks is_admin live from the DB
    so a demoted admin loses access immediately."""

    @functools.wraps(fn)
    async def wrapper(*args, **kwargs):
        uid = _session_user_id()
        if uid is None:
            return jsonify({"error": "authentication required"}), 401
        async with session_scope() as db:
            user = await db.get(User, uid)
            if user is None:
                session.pop(SESSION_KEY, None)
                return jsonify({"error": "authentication required"}), 401
            if not user.is_admin:
                return jsonify({"error": "admin access required"}), 403
        g.user_id = uid
        return await fn(*args, **kwargs)

    return wrapper


@bp.post("/register")
async def register():
    data = await request.get_json(silent=True) or {}
    email = (data.get("email") or "").strip().lower()
    password = data.get("password") or ""
    display_name = (data.get("display_name") or "").strip()

    if not email or "@" not in email:
        return jsonify({"error": "a valid email is required"}), 400
    if len(password) < MIN_PASSWORD_LEN:
        return jsonify({"error": f"password must be at least {MIN_PASSWORD_LEN} characters"}), 400
    if not display_name:
        display_name = email.split("@", 1)[0]

    async with session_scope() as db:
        user_count = await db.scalar(select(func.count()).select_from(User)) or 0
        is_first = user_count == 0
        # The first account bootstraps the admin and is always allowed, even when
        # registration is otherwise closed.
        if not is_first and not await get_setting(db, "allow_registration"):
            return jsonify({"error": "registration is closed"}), 403
        existing = await db.scalar(select(User).where(User.email == email))
        if existing is not None:
            return jsonify({"error": "an account with that email already exists"}), 409
        user = User(
            email=email,
            password_hash=hash_password(password),
            display_name=display_name,
            is_admin=is_first,
        )
        db.add(user)
        await db.commit()
        await db.refresh(user)
        session[SESSION_KEY] = str(user.id)
        session.permanent = True
        return jsonify(_serialize_user(user)), 201


@bp.post("/login")
async def login():
    data = await request.get_json(silent=True) or {}
    email = (data.get("email") or "").strip().lower()
    password = data.get("password") or ""

    async with session_scope() as db:
        user = await db.scalar(select(User).where(User.email == email))
        if user is None or not user.password_hash or not verify_password(password, user.password_hash):
            return jsonify({"error": "invalid email or password"}), 401
        session[SESSION_KEY] = str(user.id)
        session.permanent = True
        return jsonify(_serialize_user(user))


@bp.post("/logout")
async def logout():
    session.pop(SESSION_KEY, None)
    return jsonify({"ok": True})


@bp.get("/me")
@login_required
async def me():
    async with session_scope() as db:
        user = await db.get(User, g.user_id)
        if user is None:
            session.pop(SESSION_KEY, None)
            return jsonify({"error": "authentication required"}), 401
        return jsonify(_serialize_user(user))