diff --git a/gateway/app.py b/gateway/app.py index 63c6b77..241d674 100644 --- a/gateway/app.py +++ b/gateway/app.py @@ -6,6 +6,7 @@ import os from pathlib import Path from datetime import datetime, timezone from typing import Any +from urllib.parse import parse_qsl import httpx from fastapi import FastAPI, HTTPException, Request @@ -17,6 +18,9 @@ from services.shared.models import HealthResponse app = FastAPI(title="api-gateway", version="1.0.0") logger = logging.getLogger("api-gateway.proxy") _ERROR_ALERT_SENT_AT: dict[str, float] = {} +_ERROR_ALERT_MAX_MESSAGE_LENGTH = 3800 +_ERROR_ALERT_HEADER_PREVIEW_LIMIT = 120 +_ERROR_ALERT_BODY_PREVIEW_LIMIT = 1200 def _env_flag(name: str, default: bool = False) -> bool: @@ -102,6 +106,168 @@ def _error_alert_key(service: str, status: int, path: str, method: str) -> str: return f"{service}:{status}:{method}:{path}" +def _error_alert_mask_value(value: str) -> str: + text = value.strip() + if len(text) <= 6: + return "" + return f"{text[:3]}...{text[-2:]}" + + +def _error_alert_headers_preview(headers: Any) -> list[str]: + sensitive = { + "authorization", + "x-telegram-bot-api-secret-token", + "x-whatsapp-webhook-secret", + "x-hub-signature-256", + "cookie", + "set-cookie", + "proxy-authorization", + "x-api-key", + "api-key", + "x-auth-token", + "x-service-token", + } + result: list[str] = [] + for key, value in headers.items(): + key_lower = key.lower() + if key_lower in sensitive: + value = _error_alert_mask_value(value) + result.append(f"{key}: {value}") + + return result[:_ERROR_ALERT_HEADER_PREVIEW_LIMIT] + + +def _error_alert_mask_query(query: str) -> str: + if not query: + return "-" + + sensitive = { + "token", + "access_token", + "refresh_token", + "authorization", + "password", + "secret", + "api_key", + "apikey", + "x_api_key", + "x-telegram-bot-api-secret-token", + "x-whatsapp-webhook-secret", + "signature", + } + parsed = parse_qsl(query, keep_blank_values=True) + if not parsed: + return query + + values = [] + for key, value in parsed: + key_lower = key.lower() + if key_lower in sensitive: + values.append(f"{key}=") + else: + values.append(f"{key}={value}") + return "&".join(values) + + +def _error_alert_request_context(request: Request) -> dict[str, Any]: + query = request.url.query + client = request.client + client_addr = f"{client.host}:{client.port}" if client else "unknown" + + return { + "client": client_addr, + "http_version": request.scope.get("http_version", "-"), + "query": _error_alert_mask_query(query), + "user_agent": request.headers.get("user-agent", "-"), + "referer": request.headers.get("referer", "-"), + "host": request.headers.get("host", "-"), + "request_id": request.headers.get("x-request-id") + or request.headers.get("x-correlation-id") + or request.headers.get("x-amzn-trace-id") + or "-", + "x_user": request.headers.get("x-user", "-"), + "x_role": request.headers.get("x-role", "-"), + "x_tenant_id": request.headers.get("x-tenant-id", "-"), + "x_forwarded_for": request.headers.get("x-forwarded-for", "-"), + "headers_count": len(request.headers), + "headers": _error_alert_headers_preview(request.headers), + } + + +def _error_alert_message_payload( + *, + service: str, + method: str, + path: str, + status: int, + detail: str, + request: Request, + extra: dict[str, Any] | None = None, +) -> str: + timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + request_context = _error_alert_request_context(request) + + lines = [ + f"🚨 [{service}] {status}", + f"time: {timestamp}", + f"method: {method}", + f"path: {path}", + f"service: {service}", + f"client: {request_context['client']}", + f"http_version: {request_context['http_version']}", + f"query: {request_context['query']}", + f"request_id: {request_context['request_id']}", + f"user_agent: {request_context['user_agent']}", + f"referer: {request_context['referer']}", + f"host: {request_context['host']}", + f"x_user: {request_context['x_user']}", + f"x_role: {request_context['x_role']}", + f"x_tenant_id: {request_context['x_tenant_id']}", + f"x_forwarded_for: {request_context['x_forwarded_for']}", + "", + "headers:", + ] + + lines.extend(f" {item}" for item in request_context["headers"]) + + if extra: + lines.extend(("", "extra:")) + for key, value in extra.items(): + lines.append(f" {key}: {value}") + + lines.append("") + lines.append(f"detail: {detail}") + + full_message = "\n".join(lines) + if len(full_message) <= _ERROR_ALERT_MAX_MESSAGE_LENGTH: + return full_message + + cutoff = _ERROR_ALERT_MAX_MESSAGE_LENGTH - 80 + if cutoff < 100: + return full_message[:_ERROR_ALERT_MAX_MESSAGE_LENGTH] + + return f"{full_message[:cutoff]}\n... (truncated)" + + +def _error_alert_body_preview(payload: bytes | None) -> str: + if payload is None: + return "-" + if len(payload) > _ERROR_ALERT_BODY_PREVIEW_LIMIT: + payload = payload[:_ERROR_ALERT_BODY_PREVIEW_LIMIT] + truncated = True + else: + truncated = False + + try: + text = payload.decode("utf-8") + except UnicodeDecodeError: + text = payload.decode("utf-8", errors="replace") + + if truncated: + return f"{text}\n... (truncated body preview)" + return text + + def _error_alert_is_throttled(key: str) -> bool: now = datetime.now(tz=timezone.utc).timestamp() cooldown = _error_alert_cooldown_seconds() @@ -136,7 +302,15 @@ async def _send_telegram_alert(message: str) -> None: ) -def _queue_telegram_error_alert(service: str, method: str, path: str, status: int, detail: str) -> None: +def _queue_telegram_error_alert( + service: str, + method: str, + path: str, + status: int, + detail: str, + request: Request | None = None, + extra: dict[str, Any] | None = None, +) -> None: if not _error_alert_enabled(): return if not _error_alert_should_alert_path(path): @@ -147,13 +321,17 @@ def _queue_telegram_error_alert(service: str, method: str, path: str, status: in if _error_alert_is_throttled(_error_alert_key(service, status, path, method)): return - timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - message = ( - f"🚨 [{service}] {status}\n" - f"method: {method}\n" - f"path: {path}\n" - f"time: {timestamp}\n" - f"detail: {detail}" + if request is None: + return + + message = _error_alert_message_payload( + service=service, + method=method, + path=path, + status=status, + detail=detail, + request=request, + extra=extra, ) try: @@ -344,6 +522,7 @@ async def error_notification_middleware(request: Request, call_next): path=str(request.url.path), status=status_code, detail=f"{type(exc).__name__}: {exc}", + request=request, ) raise @@ -354,6 +533,11 @@ async def error_notification_middleware(request: Request, call_next): path=str(request.url.path), status=response.status_code, detail=f"HTTP {response.status_code} from {request.method} {request.url.path}", + request=request, + extra={ + "response_content_type": response.headers.get("content-type", "-"), + "response_content_length": response.headers.get("content-length", "-"), + }, ) return response @@ -574,6 +758,7 @@ async def _forward(method: str, service: str, path: str, request: Request) -> Re if payload == b"": payload = None + body_preview = "-" try: async with httpx.AsyncClient(timeout=20) as client: resp = await client.request( @@ -583,13 +768,22 @@ async def _forward(method: str, service: str, path: str, request: Request) -> Re content=payload, headers=headers, ) - except httpx.RequestError: + except httpx.RequestError as exc: + body_preview = _error_alert_body_preview(payload) _queue_telegram_error_alert( service=_error_alert_service_name(), method=method, path=f"/proxy/{service}/{path}", status=502, - detail=f"Upstream service unavailable: service={service} base_url={base}", + detail=f"{type(exc).__name__}: {exc}", + request=request, + extra={ + "upstream_service": service, + "upstream_base_url": base, + "upstream_target_path": target_path, + "upstream_url": url, + "request_body_preview": body_preview, + }, ) logger.warning( "Upstream service unavailable: service=%s base_url=%s method=%s path=%s", diff --git a/tests/test_gateway_proxy.py b/tests/test_gateway_proxy.py index 43e396b..97c8492 100644 --- a/tests/test_gateway_proxy.py +++ b/tests/test_gateway_proxy.py @@ -217,13 +217,21 @@ def test_gateway_queues_error_alert_for_proxy_unreachable(monkeypatch): alerts: list[dict] = [] - def queue_alert(service: str, method: str, path: str, status: int, detail: str) -> None: # noqa: ARG001 + def queue_alert( + service: str, + method: str, + path: str, + status: int, + detail: str, + **kwargs: object, + ) -> None: # noqa: ARG001 alerts.append({ 'service': service, 'method': method, 'path': path, 'status': status, 'detail': detail, + 'kwargs': kwargs, }) monkeypatch.setattr(DummyAsyncClient, 'request', request) @@ -237,15 +245,11 @@ def test_gateway_queues_error_alert_for_proxy_unreachable(monkeypatch): response = client.get('/proxy/sales/pipelines') assert response.status_code == 502 - assert alerts == [ - { - 'service': 'api-gateway', - 'method': 'GET', - 'path': '/proxy/sales/pipelines', - 'status': 502, - 'detail': 'Upstream service unavailable: service=sales base_url=http://localhost:8020', - }, - ] + assert alerts[0]['service'] == 'api-gateway' + assert alerts[0]['method'] == 'GET' + assert alerts[0]['path'] == '/proxy/sales/pipelines' + assert alerts[0]['status'] == 502 + assert 'All connection attempts failed' in alerts[0]['detail'] def test_gateway_queues_error_alert_for_channel_alias(monkeypatch): @@ -254,13 +258,21 @@ def test_gateway_queues_error_alert_for_channel_alias(monkeypatch): alerts: list[dict] = [] - def queue_alert(service: str, method: str, path: str, status: int, detail: str) -> None: # noqa: ARG001 + def queue_alert( + service: str, + method: str, + path: str, + status: int, + detail: str, + **kwargs: object, + ) -> None: # noqa: ARG001 alerts.append({ 'service': service, 'method': method, 'path': path, 'status': status, 'detail': detail, + 'kwargs': kwargs, }) monkeypatch.setattr(DummyAsyncClient, 'request', request) @@ -283,13 +295,21 @@ def test_gateway_queues_error_alert_for_internal_exception(monkeypatch): alerts: list[dict] = [] - def queue_alert(service: str, method: str, path: str, status: int, detail: str) -> None: # noqa: ARG001 + def queue_alert( + service: str, + method: str, + path: str, + status: int, + detail: str, + **kwargs: object, + ) -> None: # noqa: ARG001 alerts.append({ 'service': service, 'method': method, 'path': path, 'status': status, 'detail': detail, + 'kwargs': kwargs, }) monkeypatch.setattr(gateway_module, '_forward', broken_forward) @@ -309,6 +329,54 @@ def test_gateway_queues_error_alert_for_internal_exception(monkeypatch): assert alerts[0]['path'] == '/proxy/sales/pipelines' +def test_gateway_queues_error_alert_with_body_and_extra_context(monkeypatch): + async def request(self, method, url, params=None, content=None, headers=None): # noqa: ANN001, ANN201 + raise gateway_module.httpx.ConnectError("All connection attempts failed") + + alerts: list[dict] = [] + + def queue_alert( + service: str, + method: str, + path: str, + status: int, + detail: str, + request=None, + extra=None, + **kwargs: object, + ) -> None: # noqa: ARG001 + alerts.append({ + 'service': service, + 'method': method, + 'path': path, + 'status': status, + 'detail': detail, + 'request': request, + 'extra': extra, + }) + + monkeypatch.setattr(DummyAsyncClient, 'request', request) + monkeypatch.setattr(gateway_module.httpx, 'AsyncClient', DummyAsyncClient) + monkeypatch.setattr(gateway_module, '_queue_telegram_error_alert', queue_alert) + monkeypatch.setenv('ERROR_TELEGRAM_ALERTS', '1') + monkeypatch.setenv('ERROR_TELEGRAM_ALERT_BOT_TOKEN', 'token') + monkeypatch.setenv('ERROR_TELEGRAM_ALERT_CHAT_IDS', '999') + client = TestClient(gateway_module.app) + + response = client.post( + '/proxy/sales/pipelines?name=demo&token=super-secret', + headers={'Authorization': 'Bearer secret-token', 'X-Request-Id': 'req-123'}, + json={'foo': 'bar'}, + ) + + assert response.status_code == 502 + assert alerts[0]['status'] == 502 + assert alerts[0]['extra']['upstream_service'] == 'sales' + assert alerts[0]['extra']['upstream_target_path'] == 'api/v1/pipelines' + assert alerts[0]['extra']['request_body_preview'] == '{"foo":"bar"}' + assert alerts[0]['request'] is not None + + def test_gateway_normalizes_channel_alias_for_t_me(monkeypatch): monkeypatch.setenv('ERROR_TELEGRAM_ALERT_CHANNEL_ID', 't.me/konturaitelecom')