diff --git a/cgi_redirect.py b/cgi_redirect.py new file mode 100644 index 0000000..f3a00f4 --- /dev/null +++ b/cgi_redirect.py @@ -0,0 +1,40 @@ +""" +CGI Implementation for proxy redirection. +Executes the redirect logic and outputs standard CGI headers and body. +""" + +import os + +from proxy_redirect import ProxyRedirectBase + + +class CGIProxyRedirect(ProxyRedirectBase): + """ + Subclass of ProxyRedirectBase that handles CGI-specific I/O. + """ + + def __init__(self): + """ + Initializes the CGI proxy using the system environment variables. + """ + super().__init__(dict(os.environ)) + + def emit(self): + """Executes the logic and prints to STDOUT in CGI format.""" + response = self.run() + + # Emit headers + print(f"Status: {response['status']}") + for key, value in response.get("headers", {}).items(): + print(f"{key}: {value}") + + # Header/Body separator + print() + + # Emit body + print(response.get("body", "")) + + +if __name__ == "__main__": + app = CGIProxyRedirect() + app.emit() diff --git a/fast_api.py b/fast_api.py new file mode 100644 index 0000000..7250063 --- /dev/null +++ b/fast_api.py @@ -0,0 +1,42 @@ +""" +FastAPI Implementation for proxy redirection. +Maps FastAPI Request objects to the pure proxy logic and returns a FastAPI Response. +""" + +from fastapi import Request, Response + +from proxy_redirect import ProxyRedirectBase + + +class FastAPIProxyRedirect(ProxyRedirectBase): + """ + Subclass of ProxyRedirectBase that interfaces with FastAPI objects. + """ + + def __init__(self, request: Request): + """ + Transforms the FastAPI Request headers into an environment dictionary + compatible with ProxyRedirectBase. + """ + env_data = { + "BACKEND_SERVER": request.headers.get("X-Backend-Server", ""), + "AUTH_REDIRECT_LOCATION": request.headers.get("X-Auth-Redirect", ""), + "APP_REDIRECT_LOCATION": request.headers.get("X-App-Redirect", ""), + "AUTH_STATUS": request.headers.get("X-Auth-Status", ""), + "REDIRECT_STATUS": request.headers.get("X-Redirect-Status", "302"), + "REQUEST_URI": str(request.url), + "REQUEST_METHOD": request.method, + } + super().__init__(env_data) + + def get_response(self) -> Response: + """ + Executes the pure logic and returns a FastAPI-compatible Response. + """ + data = self.run() + + return Response( + content=data.get("body", ""), + status_code=data.get("status", 200), + headers=data.get("headers", {}), + ) diff --git a/main.py b/main.py new file mode 100644 index 0000000..8e595ed --- /dev/null +++ b/main.py @@ -0,0 +1,50 @@ +""" +Entry point for the redirection service. +Parses command-line arguments to execute either the CGI or FastAPI implementation. +""" + +import argparse + + +def run(): + """ + Parses arguments and initializes the selected execution mode. + """ + parser = argparse.ArgumentParser(description="Run the proxy redirect service.") + parser.add_argument( + "--mode", + choices=["cgi", "fastapi"], + required=True, + help="Execution mode: 'cgi' for standard output, 'fastapi' to launch the server.", + ) + parser.add_argument( + "--addr", + required=False, + help="The IP address to bind the FastAPI server.", + ) + parser.add_argument( + "--port", + type=int, + required=False, + help="The port number to bind the FastAPI server.", + ) + + args = parser.parse_args() + + if args.mode == "fastapi": + if args.addr is None or args.port is None: + parser.error("--addr and --port are required when --mode is fastapi") + + import uvicorn + + uvicorn.run("router:app", host=args.addr, port=args.port, reload=False) + + elif args.mode == "cgi": + from cgi_redirect import CGIProxyRedirect + + app = CGIProxyRedirect() + app.emit() + + +if __name__ == "__main__": + run() diff --git a/proxy_redirect.py b/proxy_redirect.py new file mode 100644 index 0000000..7e3e537 --- /dev/null +++ b/proxy_redirect.py @@ -0,0 +1,62 @@ +""" +Orchestration layer for proxy redirection. +Defines the base class that utilizes pure utility functions to determine redirect state. +""" + +from typing import Dict, Any + + +from utils import ( + extract_request_context, + handle_no_redirect_needed, + check_health, + output_custom_redirect, +) + + +class ProxyRedirectBase: + """ + Base orchestrator. Manages state and pure function execution. + """ + + def __init__(self, env: Dict[str, Any]): + """ + Initializes the context dictionary using the provided environment. + """ + self.context = extract_request_context(env) + + def run(self) -> Dict[str, Any]: + """ + Orchestrates the logic. Returns a dictionary representing + the final response payload. + """ + # 1. Check if we can proceed or if an immediate response is needed + intervention = handle_no_redirect_needed( + self.context["redirect"], self.context["request_uri"] + ) + if intervention: + status, body = intervention + return { + "status": status, + "headers": {"Content-Type": "text/plain"}, + "body": body, + } + + # 2. Perform health check + health_result = check_health( + self.context["skip_health_check"], + self.context["backend"], + self.context["redirect"], + self.context["redirect_status"], + ) + + # 3. If the health check dictates a redirect, generate the HTML + if health_result["action"] == "redirect": + return output_custom_redirect(health_result["url"], health_result["status"]) + + # 4. Fallback for errors + return { + "status": health_result.get("status", 500), + "headers": {"Content-Type": "text/plain"}, + "body": health_result.get("body", "Internal Server Error"), + } diff --git a/router.py b/router.py new file mode 100644 index 0000000..327792d --- /dev/null +++ b/router.py @@ -0,0 +1,20 @@ +""" +FastAPI routing configuration. +Maps incoming HTTP requests to the FastAPIProxyRedirect class. +""" + +from fastapi import FastAPI, Request + + +from fast_api import FastAPIProxyRedirect + +app = FastAPI() + + +@app.get("/{full_path:path}") +async def proxy_handler(request: Request): + """ + Catch-all route that passes the request to the proxy logic and returns the response. + """ + proxy = FastAPIProxyRedirect(request) + return proxy.get_response() diff --git a/shell.nix b/shell.nix new file mode 100644 index 0000000..7f83931 --- /dev/null +++ b/shell.nix @@ -0,0 +1,11 @@ +{pkgs ? import {}}: +pkgs.mkShell { + buildInputs = [ + (pkgs.python3.withPackages (ps: + with ps; [ + fastapi + uvicorn + requests + ])) + ]; +} diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..6352363 --- /dev/null +++ b/utils.py @@ -0,0 +1,209 @@ +# Missing docstring +import requests +from urllib.parse import urlparse +from typing import Dict, Any, Optional, Tuple +import datetime + + +def debug_log(message: str) -> None: + """ + Appends a timestamped debug message to /tmp/proxy_debug.log. + Equivalent to PHP error_log with message_type 3. + """ + timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + log_entry = f"[{timestamp}] PROXY DEBUG: {message}\n" + + try: + with open("/tmp/proxy_debug.log", "a") as f: + f.write(log_entry) + except OSError as e: + # Fallback to stderr if the file is unwritable + import sys + + print(f"Failed to write to log: {e}", file=sys.stderr) + + +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""" + + + + {title} + + + + +

{title}

+

Click below if not redirected automatically.

+

{url}

+ +""" + + return { + "status": status, + "headers": { + "Location": url, + "Content-Type": "text/html", + "Content-Length": str(len(html_body)), + }, + "body": html_body, + }