feat(auth): return to the original view after an auth-expiry redirect
CI / lint (push) Successful in 2s
CI / unit (push) Successful in 8s
CI / integration (push) Successful in 2m18s
CI / publish (push) Successful in 45s

When a session has expired, require_role now bounces to /login?next=<path> and
sends the user back there after re-login.

- middleware: safe_next_url() (same-site relative path/query only; rejects
  off-site, protocol-relative, javascript:, and the auth pages — no open
  redirect). _login_redirect() builds the next param; for HTMX requests it uses
  HX-Current-URL (the page, not the fragment) and HX-Redirect so the whole
  browser navigates instead of swapping the login page into a fragment.
- login GET/POST carry `next` (hidden field), validated, used on success for
  local + LDAP; OIDC stashes it in session across the IdP round-trip.
- login.html: hidden next field + next on the SSO link.
- tests for safe_next_url.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-17 13:05:37 -04:00
parent 2594ca517d
commit 71715c38d8
4 changed files with 102 additions and 8 deletions
+51 -2
View File
@@ -1,16 +1,65 @@
from __future__ import annotations
import functools
from quart import session, redirect, url_for, abort, g
from urllib.parse import quote, urlsplit
from quart import session, redirect, url_for, abort, g, request, Response
from steward.models.users import UserRole
_ROLE_ORDER = [UserRole.viewer, UserRole.operator, UserRole.admin]
# Paths we never bounce back to after login (would loop or make no sense).
_NO_RETURN = {"/login", "/logout", "/setup"}
def role_meets(user_role: UserRole, minimum_role: UserRole) -> bool:
"""True when user_role is at least minimum_role in the viewer<operator<admin order."""
return _ROLE_ORDER.index(user_role) >= _ROLE_ORDER.index(minimum_role)
def safe_next_url(raw: str | None) -> str | None:
"""Return a safe same-site relative path+query from `raw`, else None.
Accepts a relative path or a full URL (e.g. HTMX's HX-Current-URL) and
reduces it to path[?query]. Rejects off-site/protocol-relative targets and
the auth pages themselves, so it can't be abused as an open redirect.
"""
if not raw:
return None
parts = urlsplit(raw)
# Protocol-relative (//host/…) has a netloc but no scheme — reject it; a full
# same-site URL (HX-Current-URL) has a scheme and we keep only its path.
if parts.netloc and not parts.scheme:
return None
path = parts.path
if parts.query:
path += "?" + parts.query
if not path.startswith("/") or path.startswith("//") or path.startswith("/\\"):
return None
if path.split("?", 1)[0] in _NO_RETURN:
return None
return path
def _login_redirect():
"""Redirect an unauthenticated request to /login, preserving where to return.
For HTMX requests the destination is the page the user is on (HX-Current-URL)
and we use HX-Redirect so the whole browser navigates (not a fragment swap);
for normal requests it's the requested path.
"""
is_htmx = request.headers.get("HX-Request") == "true"
nxt = safe_next_url(request.headers.get("HX-Current-URL")) if is_htmx else None
if nxt is None:
nxt = safe_next_url(request.full_path)
login_url = url_for("auth.login")
if nxt:
login_url = f"{login_url}?next={quote(nxt, safe='/')}"
if is_htmx:
resp = Response("", status=401)
resp.headers["HX-Redirect"] = login_url
return resp
return redirect(login_url)
def require_role(minimum_role: UserRole):
"""Decorator: requires authenticated user with at least minimum_role.
Also allows access for validated share-token requests (viewer level only).
@@ -23,7 +72,7 @@ def require_role(minimum_role: UserRole):
return await f(*args, **kwargs)
user_id = session.get("user_id")
if not user_id:
return redirect(url_for("auth.login"))
return _login_redirect()
user_role = UserRole(session.get("user_role", "viewer"))
if _ROLE_ORDER.index(user_role) < _ROLE_ORDER.index(minimum_role):
abort(403)
+10 -5
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import bcrypt
from quart import Blueprint, render_template, request, redirect, url_for, session
from sqlalchemy import select, func
from steward.auth.middleware import login_user, logout_user
from steward.auth.middleware import login_user, logout_user, safe_next_url
from steward.models.users import User, UserRole
auth_bp = Blueprint("auth", __name__)
@@ -54,7 +54,8 @@ async def login():
return redirect(url_for("auth.setup"))
oidc_cfg = current_app.config.get("OIDC", {})
return await render_template("auth/login.html",
oidc_enabled=oidc_cfg.get("enabled", False))
oidc_enabled=oidc_cfg.get("enabled", False),
next_url=safe_next_url(request.args.get("next")))
@auth_bp.post("/login")
@@ -65,6 +66,7 @@ async def login_post():
username = form.get("username", "").strip()
password_str = form.get("password", "")
password = password_str.encode()
dest = safe_next_url(form.get("next")) or url_for("dashboard.index")
# Try local auth first
user = await get_user_by_username(current_app, username)
@@ -72,7 +74,7 @@ async def login_post():
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"))
return redirect(dest)
# Try LDAP fallback if enabled
ldap_cfg = current_app.config.get("LDAP", {})
@@ -86,13 +88,14 @@ async def login_post():
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"))
return redirect(dest)
oidc_cfg = current_app.config.get("OIDC", {})
return await render_template(
"auth/login.html",
error="Invalid credentials",
oidc_enabled=oidc_cfg.get("enabled", False),
next_url=safe_next_url(form.get("next")),
), 400
@@ -112,6 +115,7 @@ async def login_oidc():
nonce = generate_state()
session["oidc_state"] = state
session["oidc_nonce"] = nonce
session["oidc_next"] = safe_next_url(request.args.get("next"))
redirect_uri = url_for("auth.login_oidc_callback", _external=True)
url = build_authorize_url(
doc, oidc_cfg["client_id"], redirect_uri,
@@ -170,7 +174,8 @@ async def login_oidc_callback():
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"))
dest = safe_next_url(session.pop("oidc_next", None)) or url_for("dashboard.index")
return redirect(dest)
@auth_bp.get("/logout")
+2 -1
View File
@@ -11,7 +11,7 @@
</div>
{% endif %}
{% if oidc_enabled %}
<a href="/login/oidc" class="btn" style="width:100%;text-align:center;margin-bottom:1.25rem;display:block;">
<a href="/login/oidc{% if next_url %}?next={{ next_url | urlencode }}{% endif %}" class="btn" style="width:100%;text-align:center;margin-bottom:1.25rem;display:block;">
Sign in with SSO
</a>
<div style="display:flex;align-items:center;gap:0.75rem;margin-bottom:1.25rem;">
@@ -21,6 +21,7 @@
</div>
{% endif %}
<form method="post" action="/login">
{% if next_url %}<input type="hidden" name="next" value="{{ next_url }}">{% endif %}
<div class="form-group">
<input type="text" name="username" placeholder="Username" autofocus required>
</div>
+39
View File
@@ -0,0 +1,39 @@
"""Unit tests for safe_next_url — the auth post-login return-path guard."""
from steward.auth.middleware import safe_next_url
def test_relative_path_passes():
assert safe_next_url("/hosts/") == "/hosts/"
assert safe_next_url("/hosts/abc?range=24h") == "/hosts/abc?range=24h"
def test_full_url_reduced_to_path():
# HTMX HX-Current-URL is absolute; we keep only the same-site path+query.
assert safe_next_url("http://steward.local/d/1") == "/d/1"
assert safe_next_url("https://steward.local/hosts/x?a=1") == "/hosts/x?a=1"
def test_offsite_host_is_dropped_not_redirected():
# A crafted absolute URL can't become an open redirect — netloc is discarded.
assert safe_next_url("http://evil.com/phish") == "/phish"
def test_protocol_relative_rejected():
assert safe_next_url("//evil.com/x") is None
assert safe_next_url("/\\evil.com") is None
def test_scheme_only_rejected():
assert safe_next_url("javascript:alert(1)") is None
def test_auth_pages_not_returned_to():
assert safe_next_url("/login") is None
assert safe_next_url("/login?next=/x") is None
assert safe_next_url("/logout") is None
assert safe_next_url("/setup") is None
def test_empty_and_none():
assert safe_next_url(None) is None
assert safe_next_url("") is None