feat: harden study auth redirects

pingpong/auth.py

@@ -1,5 +1,6 @@
 from datetime import timedelta
 from typing import cast
+from urllib.parse import urlencode, urlsplit
 
 from fastapi.responses import RedirectResponse
 import jwt
@@ -71,6 +72,26 @@ def __init__(self, detail: str = "", user_id: str = ""):
         self.detail = detail
 
 
+def normalize_study_redirect(destination: str | None) -> str:
+    """Normalize redirect destinations to in-app study paths only."""
+    if not destination:
+        return "/"
+
+    redirect = destination.strip()
+    if not redirect.startswith("/"):
+        return "/"
+    if redirect.startswith("//") or redirect.startswith("/\\"):
+        return "/"
+    if any(ord(char) < 32 for char in redirect):
+        return "/"
+
+    parts = urlsplit(redirect)
+    if parts.scheme or parts.netloc:
+        return "/"
+
+    return redirect
+
+
 def decode_auth_token(token: str, nowfn: NowFn = utcnow) -> AuthToken:
     """Decodes the Auth Token.
 
@@ -141,23 +162,24 @@ def generate_auth_link(
         str: Auth Link
     """
     tok = encode_auth_token(str(user_id), expiry=expiry, nowfn=nowfn)
+    safe_redirect = normalize_study_redirect(redirect)
+    params = urlencode({"token": tok, "redirect": safe_redirect})
     if is_study:
         if is_study_admin:
-            return config.study_url(
-                f"/api/study/auth/admin?token={tok}&redirect={redirect}"
-            )
+            return config.study_url(f"/api/study/auth/admin?{params}")
         else:
-            return config.study_url(f"/api/study/auth?token={tok}&redirect={redirect}")
+            return config.study_url(f"/api/study/auth?{params}")
     raise ValueError("is_study must be True for study auth links")
 
 
 def redirect_with_session_study(
     destination: str, user_id: str, expiry: int = 86_400 * 30, nowfn: NowFn = utcnow
 ):
     """Redirect to the destination with a session token."""
+    safe_destination = normalize_study_redirect(destination)
     session_token = encode_auth_token(user_id, expiry=expiry, nowfn=nowfn)
     response = RedirectResponse(
-        config.study_url(destination) if destination.startswith("/") else destination,
+        config.study_url(safe_destination),
         status_code=303,
     )
     response.set_cookie(

pingpong/study/server.py

@@ -8,6 +8,7 @@
     decode_auth_token,
     decode_session_token,
     generate_auth_link,
+    normalize_study_redirect,
     redirect_with_session_study,
 )
 from pingpong.permission import StudyExpression
@@ -161,11 +162,12 @@ async def login_magic(body: schemas.MagicLoginRequest, request: Request):
         )
 
     nowfn = get_now_fn(request)
+    safe_forward = normalize_study_redirect(body.forward)
     magic_link = generate_auth_link(
         instructor.record_id,
         expiry=86_400,
         nowfn=nowfn,
-        redirect=body.forward,
+        redirect=safe_forward,
         is_study=True,
     )
 
@@ -204,11 +206,12 @@ async def login_as(body: schemas.LoginAsRequest, request: Request):
         raise HTTPException(status_code=404, detail="Instructor not found.")
 
     nowfn = get_now_fn(request)
+    safe_forward = normalize_study_redirect(body.forward)
     magic_link = generate_auth_link(
         f"{instructor.record_id}:{admin.record_id}",
         expiry=3_600,
         nowfn=nowfn,
-        redirect=body.forward,
+        redirect=safe_forward,
         is_study=True,
         is_study_admin=True,
     )
@@ -254,7 +257,7 @@ async def auth(request: Request):
     Returns:
         RedirectResponse: Redirect either to the SSO login endpoint or to the destination.
     """
-    dest = request.query_params.get("redirect", "/")
+    dest = normalize_study_redirect(request.query_params.get("redirect", "/"))
     stok = request.query_params.get("token")
     nowfn = get_now_fn(request)
     try:
@@ -263,7 +266,7 @@ async def auth(request: Request):
         raise HTTPException(status_code=401, detail=str(e))
     except TimeException as e:
         instructor = await get_instructor(e.user_id)
-        forward = request.query_params.get("redirect", "/")
+        forward = normalize_study_redirect(request.query_params.get("redirect", "/"))
         if instructor and instructor.academic_email:
             try:
                 await login_magic(
@@ -317,7 +320,7 @@ async def auth_admin(request: Request):
     Returns:
         RedirectResponse: Redirect either to the SSO login endpoint or to the destination.
     """
-    dest = request.query_params.get("redirect", "/")
+    dest = normalize_study_redirect(request.query_params.get("redirect", "/"))
     stok = request.query_params.get("token")
     nowfn = get_now_fn(request)
     try:
@@ -329,7 +332,7 @@ async def auth_admin(request: Request):
         instructor_id, admin_id = e.user_id.split(":")
         instructor = await get_instructor(instructor_id)
         admin = await get_admin_by_id(admin_id)
-        forward = request.query_params.get("redirect", "/")
+        forward = normalize_study_redirect(request.query_params.get("redirect", "/"))
         if instructor and instructor.academic_email and admin and admin.email:
             try:
                 await login_as(