Redirecting deep links after login

app/main.py

@@ -1,5 +1,6 @@
 from contextlib import asynccontextmanager
 from pathlib import Path
+from urllib.parse import quote
 
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
@@ -45,7 +46,15 @@ def create_app() -> FastAPI:
 
     @app.exception_handler(NotAuthenticated)
     async def _redirect_to_login(request: Request, exc: NotAuthenticated):
-        return RedirectResponse(url="/login", status_code=302)
+        # Preserve the originally requested URL so we can return there after
+        # login. Only meaningful for GET navigations (the ones users deep-link).
+        login_url = "/login"
+        if request.method == "GET" and request.url.path not in ("/", "/login"):
+            target = request.url.path
+            if request.url.query:
+                target += "?" + request.url.query
+            login_url = f"/login?next={quote(target, safe='')}"
+        return RedirectResponse(url=login_url, status_code=302)
 
     @app.get("/healthz", tags=["health"])
     async def healthz():

app/templates/login.html

@@ -9,6 +9,7 @@
     {% if ldap_enabled %}
 
       <form method="post" action="/login/ldap">
+        <input type="hidden" name="next" value="{{ next_url or '' }}">
         {% if ldap_error %}<div class="alert">{{ ldap_error }}</div>{% endif %}
         <label>Username
           <input type="text" name="username" value="{{ username or '' }}" required autofocus>
@@ -22,6 +23,7 @@
       <details class="local-login-panel" {% if local_error %}open{% endif %}>
         <summary>Use a local account instead</summary>
         <form method="post" action="/login">
+          <input type="hidden" name="next" value="{{ next_url or '' }}">
           {% if local_error %}<div class="alert">{{ local_error }}</div>{% endif %}
           <label>Email
             <input type="email" name="email" value="{{ email or '' }}" required>
@@ -36,6 +38,7 @@
     {% else %}
 
       <form method="post" action="/login">
+        <input type="hidden" name="next" value="{{ next_url or '' }}">
         {% if local_error %}<div class="alert">{{ local_error }}</div>{% endif %}
         <label>Email
           <input type="email" name="email" value="{{ email or '' }}" required autofocus>

app/web/pages.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
+from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, status
 from fastapi.responses import HTMLResponse, RedirectResponse, Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -14,6 +14,21 @@ from app.web.templates import templates
 router = APIRouter(include_in_schema=False)
 
 
+def _safe_next(next_url: str) -> str:
+    """Return next_url only if it's a safe same-site path, else "/".
+
+    Guards against open-redirect: rejects absolute URLs and protocol-relative
+    ("//host", "/\\host") targets, and never bounces back to the auth pages.
+    """
+    if (
+        next_url.startswith("/")
+        and not next_url.startswith(("//", "/\\"))
+        and not next_url.startswith(("/login", "/logout"))
+    ):
+        return next_url
+    return "/"
+
+
 def _set_auth_cookie(response: Response, access_token: str) -> None:
     response.set_cookie(
         key=settings.cookie_name,
@@ -35,9 +50,11 @@ async def dashboard(
 
 
 @router.get("/login", response_class=HTMLResponse)
-async def login_page(request: Request):
+async def login_page(request: Request, next_url: str = Query("", alias="next")):
     return templates.TemplateResponse(
-        request, "login.html", {"ldap_enabled": settings.ldap_enabled}
+        request,
+        "login.html",
+        {"ldap_enabled": settings.ldap_enabled, "next_url": _safe_next(next_url)},
     )
 
 
@@ -46,6 +63,7 @@ async def login_submit(
     request: Request,
     email: str = Form(...),
     password: str = Form(...),
+    next_url: str = Form("", alias="next"),
     db: AsyncSession = Depends(get_db),
 ):
     user = await user_crud.authenticate(db, email, password)
@@ -57,11 +75,14 @@ async def login_submit(
                 "local_error": "Incorrect email or password",
                 "email": email,
                 "ldap_enabled": settings.ldap_enabled,
+                "next_url": _safe_next(next_url),
             },
             status_code=status.HTTP_401_UNAUTHORIZED,
         )
     access, _ = await issue_token_pair(db, user)
-    response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
+    response = RedirectResponse(
+        url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER
+    )
     _set_auth_cookie(response, access)
     return response
 
@@ -71,6 +92,7 @@ async def login_ldap_submit(
     request: Request,
     username: str = Form(...),
     password: str = Form(...),
+    next_url: str = Form("", alias="next"),
     db: AsyncSession = Depends(get_db),
 ):
     if not settings.ldap_enabled:
@@ -84,11 +106,14 @@ async def login_ldap_submit(
                 "ldap_error": "Incorrect username or password",
                 "username": username,
                 "ldap_enabled": True,
+                "next_url": _safe_next(next_url),
             },
             status_code=status.HTTP_401_UNAUTHORIZED,
         )
     access, _ = await issue_token_pair(db, user)
-    response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
+    response = RedirectResponse(
+        url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER
+    )
     _set_auth_cookie(response, access)
     return response
 

tests/test_login_redirect.py

@@ -0,0 +1,73 @@
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_deep_link_redirects_to_login_with_next(client):
+    resp = await client.get("/scopes")
+    assert resp.status_code == 302
+    assert resp.headers["location"] == "/login?next=%2Fscopes"
+
+
+@pytest.mark.asyncio
+async def test_deep_link_preserves_query_string(client):
+    resp = await client.get("/scopes?description=alpha")
+    assert resp.status_code == 302
+    assert resp.headers["location"] == "/login?next=%2Fscopes%3Fdescription%3Dalpha"
+
+
+@pytest.mark.asyncio
+async def test_root_redirect_has_no_next(client):
+    resp = await client.get("/")
+    assert resp.status_code == 302
+    assert resp.headers["location"] == "/login"
+
+
+@pytest.mark.asyncio
+async def test_login_page_renders_next_field(client):
+    resp = await client.get("/login?next=/scopes")
+    assert resp.status_code == 200
+    assert 'name="next" value="/scopes"' in resp.text
+
+
+@pytest.mark.asyncio
+async def test_login_redirects_to_next_on_success(client, make_user):
+    await make_user("u@example.com")
+    resp = await client.post(
+        "/login",
+        data={"email": "u@example.com", "password": "secret123", "next": "/scopes"},
+    )
+    assert resp.status_code == 303
+    assert resp.headers["location"] == "/scopes"
+
+
+@pytest.mark.asyncio
+async def test_login_defaults_to_root_without_next(client, make_user):
+    await make_user("u2@example.com")
+    resp = await client.post(
+        "/login", data={"email": "u2@example.com", "password": "secret123"}
+    )
+    assert resp.status_code == 303
+    assert resp.headers["location"] == "/"
+
+
+@pytest.mark.asyncio
+async def test_login_rejects_unsafe_next(client, make_user):
+    await make_user("u3@example.com")
+    for bad in ["https://evil.com", "//evil.com", "/\\evil.com", "/login"]:
+        resp = await client.post(
+            "/login",
+            data={"email": "u3@example.com", "password": "secret123", "next": bad},
+        )
+        assert resp.status_code == 303
+        assert resp.headers["location"] == "/", bad
+
+
+@pytest.mark.asyncio
+async def test_failed_login_preserves_next(client, make_user):
+    await make_user("u4@example.com")
+    resp = await client.post(
+        "/login",
+        data={"email": "u4@example.com", "password": "wrong", "next": "/scopes"},
+    )
+    assert resp.status_code == 401
+    assert 'name="next" value="/scopes"' in resp.text

tests/test_scope_guide.py

@@ -238,4 +238,5 @@ async def test_scopes_page_forbidden_without_permission(client, seeded, make_use
 async def test_scopes_page_redirects_when_anonymous(client):
     resp = await client.get("/scopes")
     assert resp.status_code == 302
-    assert resp.headers["location"] == "/login"
+    # Deep link is preserved so the user lands back here after logging in.
+    assert resp.headers["location"] == "/login?next=%2Fscopes"