From 71715c38d8c53f27906b2f02cfe9fb0d492f59e4 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 17 Jun 2026 13:05:37 -0400 Subject: [PATCH] feat(auth): return to the original view after an auth-expiry redirect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a session has expired, require_role now bounces to /login?next= 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) --- steward/auth/middleware.py | 53 +++++++++++++++++++++++++++++-- steward/auth/routes.py | 15 ++++++--- steward/templates/auth/login.html | 3 +- tests/test_safe_next_url.py | 39 +++++++++++++++++++++++ 4 files changed, 102 insertions(+), 8 deletions(-) create mode 100644 tests/test_safe_next_url.py diff --git a/steward/auth/middleware.py b/steward/auth/middleware.py index 6aca001..e1b06c9 100644 --- a/steward/auth/middleware.py +++ b/steward/auth/middleware.py @@ -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= _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) diff --git a/steward/auth/routes.py b/steward/auth/routes.py index f1cd510..399b0e9 100644 --- a/steward/auth/routes.py +++ b/steward/auth/routes.py @@ -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") diff --git a/steward/templates/auth/login.html b/steward/templates/auth/login.html index 19c2745..a94f7e5 100644 --- a/steward/templates/auth/login.html +++ b/steward/templates/auth/login.html @@ -11,7 +11,7 @@ {% endif %} {% if oidc_enabled %} - + Sign in with SSO
@@ -21,6 +21,7 @@
{% endif %}
+ {% if next_url %}{% endif %}
diff --git a/tests/test_safe_next_url.py b/tests/test_safe_next_url.py new file mode 100644 index 0000000..4b6b9cd --- /dev/null +++ b/tests/test_safe_next_url.py @@ -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