from __future__ import annotations import bcrypt from quart import Blueprint, render_template, request, redirect, url_for, session from sqlalchemy import select, func from fablednetmon.auth.middleware import login_user, logout_user from fablednetmon.models.users import User, UserRole auth_bp = Blueprint("auth", __name__) async def get_user_count(app) -> int: async with app.db_sessionmaker() as db: result = await db.execute(select(func.count()).select_from(User)) return result.scalar_one() async def get_user_by_username(app, username: str) -> User | None: async with app.db_sessionmaker() as db: result = await db.execute(select(User).where(User.username == username)) return result.scalar_one_or_none() @auth_bp.get("/login") async def login(): from quart import current_app if await get_user_count(current_app) == 0: return redirect(url_for("auth.setup")) return await render_template("auth/login.html") @auth_bp.post("/login") async def login_post(): from quart import current_app form = await request.form username = form.get("username", "").strip() password = form.get("password", "").encode() user = await get_user_by_username(current_app, username) if not user or not user.is_active: return await render_template("auth/login.html", error="Invalid credentials"), 400 if not bcrypt.checkpw(password, user.password_hash.encode()): return await render_template("auth/login.html", error="Invalid credentials"), 400 login_user(user) return redirect(url_for("dashboard.index")) @auth_bp.get("/logout") async def logout(): logout_user() return redirect(url_for("auth.login")) @auth_bp.get("/setup") async def setup(): from quart import current_app if await get_user_count(current_app) > 0: return redirect(url_for("auth.login")) return await render_template("auth/setup.html") @auth_bp.post("/setup") async def setup_post(): from quart import current_app if await get_user_count(current_app) > 0: return redirect(url_for("auth.login")) form = await request.form username = form.get("username", "").strip() email = form.get("email", "").strip() password = form.get("password", "").encode() if not username or not email or not password: return await render_template("auth/setup.html", error="All fields required"), 400 pw_hash = bcrypt.hashpw(password, bcrypt.gensalt()).decode() user = User(username=username, email=email, password_hash=pw_hash, role=UserRole.admin) async with current_app.db_sessionmaker() as db: async with db.begin(): db.add(user) login_user(user) return redirect(url_for("dashboard.index"))