CVE-2025-55751

@@ -35,7 +35,7 @@
 from app.util.approve import Approve
 
 # Import middleware
-from app.util.auth_dependencies import Authentication, CurrentMember
+from app.util.auth_dependencies import Authentication, CurrentMember, verify_redirect_url, sign_redirect_url
 from app.util.database import get_session, init_db
 from app.util.discord import Discord
 
@@ -242,11 +242,9 @@ async def index(request: Request, token: Optional[str] = Cookie(None)):
 
 
 @app.get("/discord/new/")
-async def oauth_transformer(redir: str = "/join/2"):
-    # Open redirect check
-    hostname = urlparse(redir).netloc
-    if hostname != "" and hostname != Settings().http.domain:
-        redir = "/join/2"
+async def oauth_transformer(redir: str = None):
+    if not redir:
+        redir = sign_redirect_url("/join/2")
 
     oauth = OAuth2Session(
         Settings().discord.client_id,
@@ -278,13 +276,10 @@ async def oauth_transformer_new(
     session: Session = Depends(get_session),
 ):
     # Open redirect check
-    if redir == "_redir":
-        redir = redir_endpoint
-
-    hostname = urlparse(redir).netloc
-
-    if hostname != "" and hostname != Settings().http.domain:
-        redir = "/join/2"
+    if redir == "_redir" and redir_endpoint:
+        redir = verify_redirect_url(redir_endpoint)
+    else:
+        redir = verify_redirect_url(sign_redirect_url(redir))
 
     if code is None:
         return Errors.generate(

@@ -3,6 +3,7 @@
 import logging
 import time
 import uuid
+import hashlib
 from typing import Annotated, Optional
 
 from fastapi import Cookie, Depends, HTTPException, Request, status
@@ -121,7 +122,8 @@ def get_current_user(request: Request, token: Optional[str] = Cookie(None)) -> d
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
 
         # For web requests, redirect to Discord OAuth
-        raise HTTPException(status_code=status.HTTP_302_FOUND, detail="Authentication required", headers={"Location": f"/discord/new?redir={request.url.path}"})
+        redir_jwt = sign_redirect_url(request.url.path)
+        raise HTTPException(status_code=status.HTTP_302_FOUND, detail="Authentication required", headers={"Location": f"/discord/new?redir={redir_jwt}"})
 
     # Session timeout check (skip for API keys)
     if not user_jwt.get("api_key", False):
@@ -130,7 +132,8 @@ def get_current_user(request: Request, token: Optional[str] = Cookie(None)) -> d
             if request.headers.get("Authorization"):
                 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Session expired")
             else:
-                raise HTTPException(status_code=status.HTTP_302_FOUND, detail="Session expired", headers={"Location": f"/discord/new?redir={request.url.path}"})
+                redir_jwt = sign_redirect_url(request.url.path)
+                raise HTTPException(status_code=status.HTTP_302_FOUND, detail="Session expired", headers={"Location": f"/discord/new?redir={redir_jwt}"})
 
     # Set Sentry user context if enabled
     if Settings().telemetry.enable and set_user is not None:
@@ -211,3 +214,75 @@ def create_jwt(user):
             logger.error(f"JWT encode error: {encode_error}")
             raise ValueError(f"Failed to encode JWT: {encode_error}")
         return bearer
+
+
+def sign_redirect_url(url: str) -> str:
+    """
+    Sign a redirect URL to prevent tampering (CWE-601 mitigation).
+
+    Args:
+        url: The redirect URL to sign
+
+    Returns:
+        A signed token representing the URL
+    """
+
+    # Create a JWT payload with the URL and expiration
+    payload = {
+        "redirect_url": url,
+        "iat": int(time.time()),
+        "exp": int(time.time()) + 300,  # 5 minute expiration
+        "purpose": "redirect"
+    }
+
+
+    try:
+        token = jwt.encode(
+            {"alg": Settings().jwt.algorithm},
+            payload,
+            Settings().jwt.redir_key,
+        )
+        return token
+    except Exception as e:
+        logger.error(f"Failed to sign redirect URL: {e}")
+        raise ValueError("Failed to sign redirect URL")
+
+
+def verify_redirect_url(signed_url: str) -> str:
+    """
+    Verify a signed redirect URL and extract the original URL.
+
+    Args:
+        signed_url: The signed URL token
+
+    Returns:
+        The original URL if valid, otherwise "/join/2"
+    """
+
+
+    try:
+        decoded = jwt.decode(
+            signed_url,
+            Settings().jwt.redir_key,
+            algorithms=[Settings().jwt.algorithm],
+        )
+
+        payload = decoded.claims
+
+        # Verify this is a redirect token
+        if payload.get("purpose") != "redirect":
+            logger.warning("Invalid token purpose for redirect")
+            return "/join/2"
+
+        redirect_url = payload.get("redirect_url", "/join/2")
+
+        # Basic sanity check - must be relative URL
+        if redirect_url.startswith("/") and not redirect_url.startswith("//"):
+            return redirect_url
+        else:
+            logger.warning(f"Invalid redirect URL format: {redirect_url}")
+            return "/join/2"
+
+    except Exception as e:
+        logger.warning(f"Failed to verify redirect URL: {e}")
+        return "/join/2"

@@ -8,6 +8,7 @@
 import subprocess
 from typing import List, Optional
 
+from requests_oauthlib.oauth1_session import OAuth1
 import yaml
 from joserfc.jwk import OctKey
 from pydantic import BaseModel, Field, SecretStr, constr, model_validator
@@ -281,23 +282,23 @@ class JwtConfig(BaseModel):
     lifetime_user: Optional[int] = Field(9072000)
     lifetime_sudo: Optional[int] = Field(86400)
     key_object: Optional[OctKey] = Field(default=None, exclude=True)
+    redir_key: Optional[OctKey] = Field(default=None, exclude=True)
+    oauth_key: Optional[OctKey] = Field(default=None, exclude=True)
 
     def __init__(self, **data):
         super().__init__(**data)
         # Create JWT key object during initialization
         try:
             self.key_object = OctKey.import_key(self.secret.get_secret_value())
+            self.redir_key = OctKey.import_key(self.secret.get_secret_value() + "redir")
+            self.oauth_key = OctKey.import_key(self.secret.get_secret_value() + "oauth")
         except Exception as e:
             raise ValueError(f"Invalid JWT secret key: {e}") from e
 
 
 if settings.get("jwt"):
     jwt_config = JwtConfig(**settings["jwt"])
-elif onboard_env == "dev":
-    # Provides a stable secret per dev instance, horribly insecure for prod
-    hostname = socket.gethostname()
-    secret = hashlib.sha256(hostname.encode("utf-8")).hexdigest()
-    jwt_config = JwtConfig(secret=secret)
+
 
 
 class InfraConfig(BaseModel):