# Missing docstring
import requests
from urllib.parse import urlparse
from typing import Dict, Any, Optional, Tuple
import datetime
from logger import logger as debug_log
def extract_request_context(env: Dict[str, Any]) -> Dict[str, Any]:
"""
Purifies the extraction and calculation of request metadata.
Input: A raw dictionary of environment variables.
Output: A dictionary of processed, typed data ready for logic functions.
"""
# Extract raw values with defaults
backend = env.get("BACKEND_SERVER", "")
auth_redirect = env.get("AUTH_REDIRECT_LOCATION", "")
app_redirect = env.get("APP_REDIRECT_LOCATION", "")
auth_status = str(env.get("AUTH_STATUS", ""))
skip_health_check = env.get("SKIP_HEALTH_CHECK", False)
# Type casting and normalization
try:
redirect_status = int(env.get("REDIRECT_STATUS", 302))
except (ValueError, TypeError):
redirect_status = 302
if redirect_status == 0:
redirect_status = 302
request_uri = env.get("REQUEST_URI", "")
request_method = env.get("REQUEST_METHOD", "GET")
# Determine if this is an auth redirect or app redirect
is_auth_redirect = auth_status == "401"
# Logic: If 401 and auth_redirect isn't "false", use auth_redirect; else use app_redirect
redirect = (
auth_redirect
if (is_auth_redirect and auth_redirect != "false")
else app_redirect
)
# Return pure data structure
return {
"backend": backend,
"auth_redirect": auth_redirect,
"app_redirect": app_redirect,
"auth_status": auth_status,
"skip_health_check": bool(skip_health_check),
"redirect_status": redirect_status,
"request_uri": request_uri,
"request_method": request_method,
"is_auth_redirect": is_auth_redirect,
"redirect": redirect,
}
def handle_no_redirect_needed(
redirect: Optional[str], request_uri: str
) -> Optional[Tuple[int, str]]:
"""
Evaluates if a redirect is missing and determines the response.
Returns: Tuple (status_code, body) if an intervention is needed,
otherwise None to indicate 'pass through'.
"""
if not redirect:
if "/loginError" in request_uri:
debug_log("Login error page - no redirect needed, passing through")
return (200, "Login error - please try again")
debug_log("ERROR: No redirect destination available")
return (500, "Missing redirect parameter.")
return None
def check_health(
skip_health_check: bool, backend: str, redirect: str, redirect_status: int
) -> Dict[str, Any]:
"""
Validates backend health and returns a data structure for the caller to handle.
Returns: { "action": str, "status": int, "body": str, "url": Optional[str] }
"""
# Validation logic
is_valid_url = False
try:
result = urlparse(backend if "://" in backend else f"http://{backend}")
is_valid_url = all([result.scheme, result.netloc])
except ValueError:
is_valid_url = False
if not skip_health_check and backend != "/health" and not is_valid_url:
debug_log(f"ERROR: Invalid backend hostname '{backend}'")
return {
"action": "error",
"status": 500,
"body": f"Invalid backend hostname '{backend}'.",
}
if not skip_health_check and backend == "/health":
debug_log(f"Health check endpoint, redirecting to: {redirect}")
return {"action": "redirect", "status": redirect_status, "url": redirect}
debug_log(f"Checking backend health: {backend}")
http_code = 0
error_msg = None
try:
# Perform HEAD request (CURLOPT_NOBODY equivalent)
response = requests.head(
backend if "://" in backend else f"http://{backend}",
timeout=2,
allow_redirects=False,
)
http_code = response.status_code
except requests.RequestException as e:
error_msg = str(e)
debug_log(f"CURL error (Requests): {error_msg}")
debug_log(f"Backend response code: {http_code}")
if http_code == 200:
debug_log(f"Backend healthy (code: {http_code}), redirecting to: {redirect}")
return {"action": "redirect", "status": redirect_status, "url": redirect}
else:
debug_log(f"Backend unavailable (code: {http_code}), returning 503")
return {
"action": "error",
"status": 503,
"body": "Backend unavailable.",
}
# Unconditional redirect (current logic)
return {"action": "redirect", "status": redirect_status, "url": redirect}
# I don't know what this conditional logic was for... maybe to see if the server was down?
# I'll add it back in when it matters
# healthy_codes = [200, 301, 302, 403]
# if http_code in healthy_codes:
# debug_log(f"Backend healthy (code: {http_code}), redirecting to: {redirect}")
# return {"action": "redirect", "status": redirect_status, "url": redirect}
# else:
# debug_log(f"Backend unhealthy (code: {http_code}), returning 503")
# return {"action": "render", "status": 503, "template": "errors/503.html"}
def output_custom_redirect(url: str, status: int) -> Dict[str, Any]:
"""
Purified: Generates redirect metadata and HTML body.
Returns: A dictionary containing headers and the HTML string.
"""
debug_log(f"Generating redirect data: {status} -> {url}")
titles = {301: "Moved Permanently", 302: "Found"}
title = f"{status} {titles.get(status, 'Redirect')}"
styles = """
body { font-family: sans-serif; text-align: center; margin-top: 10%; }
a { color: blue; text-decoration: underline; }
"""
html_body = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>{title}</title>
<meta http-equiv="refresh" content="0; url={url}">
<style type="text/css">
{styles}
</style>
</head>
<body>
<h1>{title}</h1>
<p>Click below if not redirected automatically.</p>
<p><a href="{url}">{url}</a></p>
</body>
</html>"""
return {
"status": status,
"headers": {
"Location": url,
"Content-Type": "text/html",
"Content-Length": str(len(html_body)),
},
"body": html_body,
}