38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
|
"""Auth helpers and decorators."""
|
||
|
|
from functools import wraps
|
||
|
|
from flask import abort, redirect, url_for, session, flash, request
|
||
|
|
from flask_login import current_user
|
||
|
|
|
||
|
|
|
||
|
|
def admin_required(view):
|
||
|
|
@wraps(view)
|
||
|
|
def wrapped(*args, **kwargs):
|
||
|
|
if not current_user.is_authenticated:
|
||
|
|
return redirect(url_for("auth.login", next=request.full_path))
|
||
|
|
if not current_user.is_admin:
|
||
|
|
abort(403)
|
||
|
|
if session.get("mfa_pending"):
|
||
|
|
return redirect(url_for("auth.mfa_challenge"))
|
||
|
|
return view(*args, **kwargs)
|
||
|
|
return wrapped
|
||
|
|
|
||
|
|
|
||
|
|
def login_required_full(view):
|
||
|
|
"""Like flask_login.login_required, but also blocks users with pending MFA."""
|
||
|
|
@wraps(view)
|
||
|
|
def wrapped(*args, **kwargs):
|
||
|
|
if not current_user.is_authenticated:
|
||
|
|
return redirect(url_for("auth.login", next=request.full_path))
|
||
|
|
if session.get("mfa_pending"):
|
||
|
|
return redirect(url_for("auth.mfa_challenge"))
|
||
|
|
return view(*args, **kwargs)
|
||
|
|
return wrapped
|
||
|
|
|
||
|
|
|
||
|
|
def get_remote_ip() -> str:
|
||
|
|
"""Best-effort IP detection for audit logs."""
|
||
|
|
fwd = request.headers.get("X-Forwarded-For", "")
|
||
|
|
if fwd:
|
||
|
|
return fwd.split(",")[0].strip()
|
||
|
|
return request.remote_addr or ""
|