from __future__ import annotations from typing import Dict from urllib.parse import quote import httpx from fastapi import APIRouter, Request from fastapi.responses import JSONResponse, StreamingResponse from starlette.background import BackgroundTask from app.config import get_settings from app.core.logger import get_logger from app.utils.http_utils import SharedAsyncClient logger = get_logger(__name__) _settings = get_settings() router = APIRouter(tags=["WhatsApp"]) # RFC 7230 hop-by-hop headers. They are meaningless when a gateway relays a # request to another service and must never be forwarded. _HOP_BY_HOP_HEADERS = frozenset({ "connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade", }) # Request headers that must never reach the internal WhatsApp service: # hop-by-hop headers plus Host (httpx sets it from the target URL), length # headers (httpx recomputes them from the body), content-encoding (httpx # transparently decompresses responses), and the gateway's own credentials # (Authorization / cookie) which belong to the main API, not the backend. _BLOCKED_REQUEST_HEADERS = _HOP_BY_HOP_HEADERS | { "host", "content-length", "accept-encoding", "authorization", "cookie", } # Response headers preserved when relaying the upstream reply back to the # client. Everything else (server version headers, content-encoding, hop-by-hop # headers, ...) is dropped so internal implementation details never leak. _RESPONSE_HEADERS_ALLOWLIST = frozenset({ "content-type", "content-disposition", "content-language", "cache-control", "etag", "expires", "last-modified", "location", "retry-after", "www-authenticate", "x-request-id", "x-correlation-id", "content-range", "accept-ranges", }) _ALLOWED_METHODS = ("GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD") # Characters preserved while percent-encoding the forwarded path so a client # supplied path can never produce a malformed upstream URL. _PATH_SAFE_CHARS = "/:@!$&'()*+,;=~-_." def _build_proxy_client() -> SharedAsyncClient: """Shared, lazily-created httpx client for upstream WhatsApp calls.""" timeout = httpx.Timeout( timeout=_settings.whatsapp_service_timeout, connect=_settings.whatsapp_service_connect_timeout, ) return SharedAsyncClient(timeout=timeout, follow_redirects=False) _proxy_client = _build_proxy_client() async def close_whatsapp_proxy_client() -> None: """Close the shared upstream client. Called once on application shutdown.""" await _proxy_client.close() def _forward_request_headers(request: Request, body: bytes) -> Dict[str, str]: """Build the header set forwarded to the WhatsApp service. Keeps application headers (content-type, accept, apikey, x-*, ...) while stripping hop-by-hop / gateway-credential headers. The `apikey` header the WhatsApp service authenticates with is taken from the client request when present (per-instance token) and falls back to the configured global key. """ headers: Dict[str, str] = {} for name, value in request.headers.items(): if name.lower() in _BLOCKED_REQUEST_HEADERS: continue headers[name] = value if body: headers.setdefault("content-type", "application/octet-stream") client_apikey = request.headers.get("apikey", "") if client_apikey: headers["apikey"] = client_apikey elif _settings.whatsapp_service_global_api_key: headers["apikey"] = _settings.whatsapp_service_global_api_key return headers def _filter_response_headers(headers: httpx.Headers) -> Dict[str, str]: return { name: value for name, value in headers.items() if name.lower() in _RESPONSE_HEADERS_ALLOWLIST } def _unavailable_response(status_code: int, detail: str) -> JSONResponse: """Sanitized gateway error response — never leaks upstream hostnames/traces.""" return JSONResponse( status_code=status_code, content={"success": False, "detail": detail}, ) @router.api_route( "/{path:path}", methods=list(_ALLOWED_METHODS), summary="Forward a request to the internal WhatsApp service", description=( "Proxies any HTTP request under /api/whatsapp to the corresponding " "endpoint of the internal WhatsApp service, preserving the HTTP method, " "path, query string, request body and relevant headers. The upstream " "response (status code and body) is returned unchanged." ), ) async def proxy_to_whatsapp(request: Request, path: str): if not _settings.whatsapp_service_enabled: return _unavailable_response(503, "WhatsApp service is not enabled") base_url = _settings.whatsapp_service_url.rstrip("/") encoded_path = quote(path, safe=_PATH_SAFE_CHARS).lstrip("/") target = f"{base_url}/{encoded_path}" if request.url.query: target = f"{target}?{request.url.query}" body = await request.body() headers = _forward_request_headers(request, body) logger.info( "Proxying %s %s -> %s (apikey=%s)", request.method, request.url.path, target, "yes" if headers.get("apikey") else "no", ) try: client = await _proxy_client.get() upstream = await client.send( client.build_request( request.method, target, content=body or None, headers=headers, ), stream=True, ) except httpx.TimeoutException as exc: logger.error("WhatsApp service timed out: %s %s: %s", request.method, target, exc) return _unavailable_response(504, "WhatsApp service timed out") except httpx.HTTPError as exc: logger.error("WhatsApp service unreachable: %s %s: %s", request.method, target, exc) return _unavailable_response(502, "WhatsApp service is unavailable") except Exception: logger.exception("Unexpected gateway error proxying %s %s", request.method, target) return _unavailable_response(502, "WhatsApp gateway error") return StreamingResponse( upstream.aiter_bytes(), status_code=upstream.status_code, headers=_filter_response_headers(upstream.headers), media_type=None, background=BackgroundTask(upstream.aclose), ) async def get_whatsapp_health() -> Dict[str, object]: """Non-fatal liveness probe used by the main application's /health endpoint. Never raises — a degraded/unreachable WhatsApp service must not take the gateway's own health check down with it. """ if not _settings.whatsapp_service_enabled: return {"configured": False, "reachable": False} url = f"{_settings.whatsapp_service_url.rstrip('/')}/server/ok" try: client = await _proxy_client.get() resp = await client.get( url, timeout=_settings.whatsapp_service_connect_timeout, ) except (httpx.HTTPError, httpx.TimeoutException) as exc: logger.warning("WhatsApp service health check failed: %s", exc) return {"configured": True, "reachable": False, "status": "unreachable"} if resp.status_code == 200: return {"configured": True, "reachable": True, "status": "ok"} return { "configured": True, "reachable": True, "status": f"unhealthy (HTTP {resp.status_code})", }