75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
"""Web-UI login via Authentik (OIDC / OpenID Connect), using Authlib.
|
|
|
|
If the OIDC_* env vars are not configured, authentication is disabled and the
|
|
UI is open (useful behind a reverse proxy that already enforces auth).
|
|
"""
|
|
import functools
|
|
|
|
from authlib.integrations.flask_client import OAuth
|
|
from flask import Blueprint, redirect, request, session, url_for
|
|
|
|
from . import settings
|
|
|
|
bp = Blueprint("auth", __name__)
|
|
_oauth = OAuth()
|
|
_client = None
|
|
|
|
|
|
def init_app(app):
|
|
"""Register the OIDC client on the Flask app (no-op if not configured)."""
|
|
global _client
|
|
_oauth.init_app(app)
|
|
if settings.oidc_enabled():
|
|
_client = _oauth.register(
|
|
name="authentik",
|
|
client_id=settings.OIDC_CLIENT_ID,
|
|
client_secret=settings.OIDC_CLIENT_SECRET,
|
|
server_metadata_url=settings.OIDC_ISSUER.rstrip("/")
|
|
+ "/.well-known/openid-configuration",
|
|
client_kwargs={"scope": "openid profile email"},
|
|
)
|
|
app.register_blueprint(bp)
|
|
|
|
|
|
def login_required(view):
|
|
@functools.wraps(view)
|
|
def wrapped(*args, **kwargs):
|
|
if not settings.oidc_enabled():
|
|
return view(*args, **kwargs) # auth disabled -> open
|
|
if not session.get("user"):
|
|
session["next"] = request.url
|
|
return redirect(url_for("auth.login"))
|
|
return view(*args, **kwargs)
|
|
|
|
return wrapped
|
|
|
|
|
|
def current_user():
|
|
return session.get("user")
|
|
|
|
|
|
@bp.route("/login")
|
|
def login():
|
|
if not settings.oidc_enabled():
|
|
return redirect(url_for("main.dashboard"))
|
|
redirect_uri = settings.OIDC_REDIRECT_URI or url_for("auth.callback", _external=True)
|
|
return _client.authorize_redirect(redirect_uri)
|
|
|
|
|
|
@bp.route("/callback")
|
|
def callback():
|
|
token = _client.authorize_access_token()
|
|
userinfo = token.get("userinfo") or _client.userinfo()
|
|
session["user"] = {
|
|
"name": userinfo.get("name") or userinfo.get("preferred_username"),
|
|
"email": userinfo.get("email"),
|
|
}
|
|
nxt = session.pop("next", None)
|
|
return redirect(nxt or url_for("main.dashboard"))
|
|
|
|
|
|
@bp.route("/logout")
|
|
def logout():
|
|
session.pop("user", None)
|
|
return redirect(url_for("main.dashboard"))
|