purewhiter--mobilegym/scripts/server/api_gateway.py

#!/usr/bin/env python3
"""
API gateway backend for mobile-gym (Nginx companion).

Feature parity with Vite apiGatewayPlugin:
- POST /api/gw/fetch  — JSON body CORS proxy (cookie jar + streaming)
- *    /api/gw/proxy  — Transparent streaming proxy (header forwarding + retry)
- Per-session cookie jar (x-gw-session header)
- 404 response cache for proxy GET (5 min TTL, max 500 entries)
- Force accept-encoding: identity to upstream

Usage (single worker, dev):
    python scripts/server/api_gateway.py --port 4181

Usage (multi-worker, production):
    uvicorn scripts.server.api_gateway:app --host 127.0.0.1 --port 4181 --workers 8
"""

import argparse
import asyncio
import json
import time
from collections import OrderedDict
from http.cookies import SimpleCookie
from urllib.parse import urlparse

import httpx
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse, Response, StreamingResponse
from starlette.routing import Route

# ── Per-worker HTTP client (created lazily) ───────────────────────────────

_client: httpx.AsyncClient | None = None


async def get_client() -> httpx.AsyncClient:
    global _client
    if _client is None or _client.is_closed:
        _client = httpx.AsyncClient(
            timeout=httpx.Timeout(30.0, connect=5.0),
            follow_redirects=True,
            limits=httpx.Limits(
                max_connections=100,
                max_keepalive_connections=20,
            ),
        )
    return _client


# ── Cookie Jar (per session, per worker) ──────────────────────────────────

_cookie_jars: dict[str, dict[str, dict[str, dict]]] = {}
# session_id -> host -> cookie_name -> {value, expires_at?}


def _get_session_id(request: Request) -> str:
    return request.headers.get("x-gw-session", "anon")


def _store_cookies(session_id: str, host: str, set_cookies: list[str]):
    if not set_cookies:
        return
    jar = _cookie_jars.setdefault(session_id, {})
    host_map = jar.setdefault(host, {})
    now = time.time()
    for sc in set_cookies:
        cookie = SimpleCookie(sc)
        for name, morsel in cookie.items():
            expires_at = None
            if morsel["max-age"]:
                try:
                    expires_at = now + int(morsel["max-age"])
                except ValueError:
                    pass
            if expires_at is not None and expires_at <= now:
                host_map.pop(name, None)
            else:
                host_map[name] = {"value": morsel.value, "expires_at": expires_at}


def _build_cookie_header(session_id: str, host: str) -> str | None:
    jar = _cookie_jars.get(session_id)
    if not jar:
        return None
    host_map = jar.get(host)
    if not host_map:
        return None
    now = time.time()
    pairs = []
    expired = []
    for name, entry in host_map.items():
        if entry.get("expires_at") is not None and entry["expires_at"] <= now:
            expired.append(name)
            continue
        pairs.append(f"{name}={entry['value']}")
    for name in expired:
        del host_map[name]
    return "; ".join(pairs) if pairs else None


# ── Proxy 404 Cache ──────────────────────────────────────────────────────

_PROXY_404_MAX = 500
_PROXY_404_TTL = 300  # 5 min

_proxy_404_cache: OrderedDict[str, float] = OrderedDict()


def _proxy_404_get(url: str) -> bool:
    expires_at = _proxy_404_cache.get(url)
    if expires_at is None:
        return False
    if time.time() > expires_at:
        _proxy_404_cache.pop(url, None)
        return False
    return True


def _proxy_404_set(url: str):
    _proxy_404_cache[url] = time.time() + _PROXY_404_TTL
    while len(_proxy_404_cache) > _PROXY_404_MAX:
        _proxy_404_cache.popitem(last=False)


# ── Hop-by-hop headers to strip ──────────────────────────────────────────

_SKIP_REQUEST_HEADERS = frozenset({
    "connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
    "te", "trailer", "transfer-encoding", "upgrade", "host",
    "origin", "referer",
    "sec-fetch-dest", "sec-fetch-mode", "sec-fetch-site", "sec-fetch-user",
    "x-gw-session",
})

_FORWARD_RESPONSE_HEADERS = frozenset({
    "content-type", "cache-control", "expires", "etag",
    "last-modified", "vary",
})

_TRANSIENT_ERRORS = frozenset({
    "ConnectError", "RemoteProtocolError",
    "ReadError", "WriteError",
})


# ── Handler: POST /api/gw/fetch ──────────────────────────────────────────

async def handle_fetch(request: Request) -> Response:
    """POST /api/gw/fetch — JSON body CORS proxy with cookie jar."""
    if request.method != "POST":
        return JSONResponse({"error": "Use POST"}, status_code=405)

    try:
        payload = await request.json()
    except Exception:
        return JSONResponse({"error": "Invalid JSON"}, status_code=400)

    target = payload.get("url")
    if not target or not isinstance(target, str):
        return JSONResponse({"error": "Missing payload.url"}, status_code=400)

    parsed = urlparse(target)
    if parsed.scheme not in ("http", "https"):
        return JSONResponse(
            {"error": "Only http/https protocols are allowed"}, status_code=400
        )

    session_id = _get_session_id(request)
    method = (payload.get("method") or "GET").upper()
    headers = payload.get("headers") or {}
    body = payload.get("body")

    # Force identity encoding
    header_keys_lower = {k.lower() for k in headers}
    if "accept-encoding" not in header_keys_lower:
        headers["accept-encoding"] = "identity"

    # Cookie jar: inject stored cookies if caller didn't set Cookie
    if "cookie" not in header_keys_lower:
        cookie = _build_cookie_header(session_id, parsed.netloc)
        if cookie:
            headers["cookie"] = cookie

    client = await get_client()
    try:
        resp = await client.request(
            method, target, headers=headers,
            content=body.encode() if isinstance(body, str) else body,
        )

        # Store Set-Cookie
        set_cookies = resp.headers.get_list("set-cookie")
        if set_cookies:
            _store_cookies(session_id, parsed.netloc, set_cookies)

        # Build response headers
        resp_headers: dict[str, str] = {}
        for h in _FORWARD_RESPONSE_HEADERS:
            v = resp.headers.get(h)
            if v:
                resp_headers[h] = v

        return Response(
            content=resp.content,
            status_code=resp.status_code,
            headers=resp_headers,
        )
    except Exception as e:
        etype = type(e).__name__
        msg = str(e)
        print(f"[gw/fetch] upstream error: {etype}: {msg} url={target}")
        return JSONResponse({"error": msg, "type": etype, "url": target}, status_code=502)


# ── Handler: /api/gw/proxy ───────────────────────────────────────────────

async def handle_proxy(request: Request) -> Response:
    """GET/POST /api/gw/proxy?url=... — streaming proxy with header forwarding."""
    target = request.query_params.get("url")
    if not target:
        return JSONResponse({"error": "Missing url param"}, status_code=400)

    parsed = urlparse(target)
    if parsed.scheme not in ("http", "https"):
        return JSONResponse(
            {"error": "Only http/https protocols are allowed"}, status_code=400
        )

    method = request.method
    if method == "GET" and _proxy_404_get(target):
        return Response(status_code=404)

    session_id = _get_session_id(request)

    # Build upstream headers: forward incoming, strip hop-by-hop
    upstream_headers: dict[str, str] = {}
    for k, v in request.headers.items():
        if k.lower() not in _SKIP_REQUEST_HEADERS:
            upstream_headers[k.lower()] = v
    upstream_headers["accept-encoding"] = "identity"

    # Cookie jar
    if "cookie" not in upstream_headers:
        cookie = _build_cookie_header(session_id, parsed.netloc)
        if cookie:
            upstream_headers["cookie"] = cookie

    # Read request body for non-GET/HEAD
    req_body = None
    if method not in ("GET", "HEAD"):
        req_body = await request.body()

    client = await get_client()

    async def do_fetch():
        return await client.request(
            method, target, headers=upstream_headers, content=req_body,
        )

    # Try with one retry for transient errors
    try:
        try:
            resp = await do_fetch()
        except httpx.HTTPError as e:
            etype = type(e).__name__
            if etype in _TRANSIENT_ERRORS:
                print(f"[gw/proxy] upstream failed, retrying: {etype}: {e}")
                await asyncio.sleep(0.12)
                resp = await do_fetch()
            else:
                raise
    except Exception as e:
        etype = type(e).__name__
        msg = str(e)
        print(f"[gw/proxy] upstream error: {etype}: {msg} url={target}")
        return JSONResponse(
            {"error": msg, "type": etype, "url": target}, status_code=502
        )

    if method == "GET" and resp.status_code == 404:
        _proxy_404_set(target)

    # Store Set-Cookie
    set_cookies = resp.headers.get_list("set-cookie")
    if set_cookies:
        _store_cookies(session_id, parsed.netloc, set_cookies)

    resp_headers: dict[str, str] = {}
    has_encoding = resp.headers.get("content-encoding")
    for h in _FORWARD_RESPONSE_HEADERS:
        v = resp.headers.get(h)
        if v:
            resp_headers[h] = v
    # Only forward content-length when no content-encoding
    if not has_encoding:
        cl = resp.headers.get("content-length")
        if cl:
            resp_headers["content-length"] = cl

    return Response(
        content=resp.content,
        status_code=resp.status_code,
        headers=resp_headers,
    )


# ── App Setup ─────────────────────────────────────────────────────────────

async def on_shutdown():
    global _client
    if _client and not _client.is_closed:
        await _client.aclose()
    _client = None


app = Starlette(
    routes=[
        Route("/api/gw/fetch", handle_fetch, methods=["POST"]),
        Route("/api/gw/proxy", handle_proxy, methods=["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"]),
    ],
    on_shutdown=[on_shutdown],
)


def main():
    import uvicorn
    parser = argparse.ArgumentParser()
    parser.add_argument("--port", type=int, default=4174)
    parser.add_argument("--workers", type=int, default=1)
    args = parser.parse_args()

    print(f"[api_gateway] API gateway on :{args.port} (workers={args.workers})")
    uvicorn.run(
        "scripts.server.api_gateway:app",
        host="127.0.0.1",
        port=args.port,
        workers=args.workers,
        log_level="warning",
    )


if __name__ == "__main__":
    main()