Newer
Older
python_redirect / utils.py
# 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,
    }