feat: OIDC/LDAP auth, audit log, maintenance windows, migrations

- Add OIDC (OpenID Connect) authentication with discovery URL support,
  group-based role mapping, and token introspection
- Add LDAP authentication with bind credentials, group-based role
  mapping, and configurable attribute names
- Add audit log (model, routes, templates) tracking settings changes,
  plugin enable/disable, login events
- Add migrations 0009 (widget variants), 0010 (maintenance windows),
  0011 (audit log)
- Add optional dependency groups to pyproject.toml: [ldap], [snmp]
- Add Settings → Auth tab with OIDC and LDAP configuration forms

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-23 08:15:15 -04:00
parent 484da1f263
commit 3c70ac56b3
15 changed files with 863 additions and 7 deletions
+139 -6
View File
@@ -8,6 +8,33 @@ from fabledscryer.models.users import User, UserRole
auth_bp = Blueprint("auth", __name__)
async def _provision_external_user(app, username: str, email: str, role_str: str):
"""Create or update a user from an external auth provider (OIDC/LDAP).
External users get a randomised unusable password so local login is blocked.
Their role is updated on every login to reflect current IdP group membership.
"""
import secrets
from fabledscryer.models.users import UserRole
async with app.db_sessionmaker() as db:
result = await db.execute(select(User).where(User.username == username))
user = result.scalar_one_or_none()
new_role = UserRole(role_str)
async with db.begin():
if user is None:
unusable_hash = bcrypt.hashpw(secrets.token_bytes(32), bcrypt.gensalt()).decode()
user = User(username=username, email=email or f"{username}@external",
password_hash=unusable_hash, role=new_role)
db.add(user)
else:
if user.role != new_role:
user.role = new_role
# Re-fetch committed user
async with app.db_sessionmaker() as db:
result = await db.execute(select(User).where(User.username == username))
return result.scalar_one()
async def get_user_count(app) -> int:
async with app.db_sessionmaker() as db:
result = await db.execute(select(func.count()).select_from(User))
@@ -25,21 +52,124 @@ 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")
oidc_cfg = current_app.config.get("OIDC", {})
return await render_template("auth/login.html",
oidc_enabled=oidc_cfg.get("enabled", False))
@auth_bp.post("/login")
async def login_post():
from quart import current_app
from fabledscryer.core.audit import log_audit
form = await request.form
username = form.get("username", "").strip()
password = form.get("password", "").encode()
password_str = form.get("password", "")
password = password_str.encode()
# Try local auth first
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
if user and user.is_active and bcrypt.checkpw(password, user.password_hash.encode()):
login_user(user)
await log_audit(current_app, user.id, user.username, "auth.login",
detail={"method": "local", "role": user.role.value})
return redirect(url_for("dashboard.index"))
# Try LDAP fallback if enabled
ldap_cfg = current_app.config.get("LDAP", {})
if ldap_cfg.get("enabled"):
from fabledscryer.auth.ldap_auth import ldap_authenticate
ldap_info = await ldap_authenticate(ldap_cfg, username, password_str)
if ldap_info:
user = await _provision_external_user(
current_app, ldap_info["username"], ldap_info["email"], ldap_info["role"]
)
login_user(user)
await log_audit(current_app, user.id, user.username, "auth.login",
detail={"method": "ldap", "role": user.role.value})
return redirect(url_for("dashboard.index"))
oidc_cfg = current_app.config.get("OIDC", {})
return await render_template(
"auth/login.html",
error="Invalid credentials",
oidc_enabled=oidc_cfg.get("enabled", False),
), 400
@auth_bp.get("/login/oidc")
async def login_oidc():
from quart import current_app
from fabledscryer.auth.oidc import get_discovery, build_authorize_url, generate_state
oidc_cfg = current_app.config.get("OIDC", {})
if not oidc_cfg.get("enabled") or not oidc_cfg.get("discovery_url"):
return redirect(url_for("auth.login"))
try:
doc = await get_discovery(oidc_cfg["discovery_url"])
except Exception as exc:
return await render_template("auth/login.html", error=f"OIDC discovery failed: {exc}",
oidc_enabled=True), 502
state = generate_state()
nonce = generate_state()
session["oidc_state"] = state
session["oidc_nonce"] = nonce
redirect_uri = url_for("auth.login_oidc_callback", _external=True)
url = build_authorize_url(
doc, oidc_cfg["client_id"], redirect_uri,
oidc_cfg.get("scopes", "openid profile email"), state, nonce,
)
return redirect(url)
@auth_bp.get("/login/oidc/callback")
async def login_oidc_callback():
from quart import current_app
from fabledscryer.auth.oidc import get_discovery, exchange_code, get_userinfo, map_role
from fabledscryer.core.audit import log_audit
oidc_cfg = current_app.config.get("OIDC", {})
error = request.args.get("error")
if error:
desc = request.args.get("error_description", error)
return await render_template("auth/login.html", error=f"SSO error: {desc}",
oidc_enabled=True), 400
code = request.args.get("code", "")
state = request.args.get("state", "")
if not code or state != session.pop("oidc_state", None):
return await render_template("auth/login.html", error="Invalid SSO state — try again.",
oidc_enabled=True), 400
try:
doc = await get_discovery(oidc_cfg["discovery_url"])
redirect_uri = url_for("auth.login_oidc_callback", _external=True)
tokens = await exchange_code(
doc, oidc_cfg["client_id"], oidc_cfg["client_secret"], code, redirect_uri
)
userinfo = await get_userinfo(doc, tokens["access_token"])
except Exception as exc:
return await render_template("auth/login.html", error=f"SSO login failed: {exc}",
oidc_enabled=True), 502
username_claim = oidc_cfg.get("username_claim", "preferred_username")
email_claim = oidc_cfg.get("email_claim", "email")
groups_claim = oidc_cfg.get("groups_claim", "groups")
username = userinfo.get(username_claim) or userinfo.get("sub", "")
email = userinfo.get(email_claim, "") or f"{username}@oidc"
groups = userinfo.get(groups_claim, [])
if not isinstance(groups, list):
groups = []
role = map_role(groups, oidc_cfg.get("admin_group", ""), oidc_cfg.get("operator_group", ""))
if not username:
return await render_template("auth/login.html",
error="SSO returned no username.",
oidc_enabled=True), 400
user = await _provision_external_user(current_app, username, email, role)
login_user(user)
await log_audit(current_app, user.id, user.username, "auth.login",
detail={"method": "oidc", "role": role})
return redirect(url_for("dashboard.index"))
@@ -60,6 +190,7 @@ async def setup():
@auth_bp.post("/setup")
async def setup_post():
from quart import current_app
from fabledscryer.core.audit import log_audit
if await get_user_count(current_app) > 0:
return redirect(url_for("auth.login"))
form = await request.form
@@ -74,4 +205,6 @@ async def setup_post():
async with db.begin():
db.add(user)
login_user(user)
await log_audit(current_app, user.id, user.username, "auth.setup",
detail={"username": username})
return redirect(url_for("dashboard.index"))