feat: session auth, role middleware, login/setup/logout routes

2026-03-17 19:06:06 -04:00
parent 5c25953929
commit 6ca25e94a0
6 changed files with 134 additions and 1 deletions
@@ -1,3 +1,77 @@
-from quart import Blueprint
+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"))
@@ -0,0 +1 @@
+{% extends "base.html" %}{% block content %}login{% endblock %}
@@ -0,0 +1 @@
+{% extends "base.html" %}{% block content %}setup{% endblock %}
@@ -0,0 +1 @@
+<!DOCTYPE html><html><body>{% block content %}{% endblock %}</body></html>
@@ -0,0 +1 @@
+{% extends "base.html" %}{% block content %}dashboard{% endblock %}
@@ -0,0 +1,55 @@
+import pytest
+import bcrypt
+from unittest.mock import AsyncMock, patch
+from fablednetmon.models.users import User, UserRole
+
+
+def make_user(role=UserRole.admin):
+    return User(
+        id="test-id",
+        username="admin",
+        email="admin@example.com",
+        password_hash=bcrypt.hashpw(b"password", bcrypt.gensalt()).decode(),
+        role=role,
+        is_active=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_login_page_returns_200(client):
+    response = await client.get("/login")
+    assert response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_login_redirects_to_setup_when_no_users(client, app):
+    with patch("fablednetmon.auth.routes.get_user_count", new=AsyncMock(return_value=0)):
+        response = await client.get("/login")
+    assert response.status_code == 302
+    assert "/setup" in response.headers["Location"]
+
+
+@pytest.mark.asyncio
+async def test_login_success_redirects_to_dashboard(client, app):
+    user = make_user()
+    with patch("fablednetmon.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
+         patch("fablednetmon.auth.routes.get_user_by_username", new=AsyncMock(return_value=user)):
+        response = await client.post("/login", form={"username": "admin", "password": "password"})
+    assert response.status_code == 302
+    assert "/" in response.headers["Location"]
+
+
+@pytest.mark.asyncio
+async def test_login_failure_returns_400(client, app):
+    user = make_user()
+    with patch("fablednetmon.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
+         patch("fablednetmon.auth.routes.get_user_by_username", new=AsyncMock(return_value=user)):
+        response = await client.post("/login", form={"username": "admin", "password": "wrong"})
+    assert response.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_require_role_redirects_unauthenticated(client):
+    response = await client.get("/")
+    assert response.status_code == 302
+    assert "/login" in response.headers["Location"]
				`@@ -0,0 +1 @@`
				`{% extends "base.html" %}{% block content %}login{% endblock %}`
				`@@ -0,0 +1 @@`
				`<!DOCTYPE html><html><body>{% block content %}{% endblock %}</body></html>`