From 2a268b3a33201290a89f0a1bb4cd6dd288ee35e1 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Thu, 30 Apr 2026 18:20:17 +0200 Subject: [PATCH] feat(V2): build_app con swagger /apidocs + middleware + handlers Aggiunge /docs e /redoc alla whitelist auth (path disabilitati, nessun rischio sicurezza). Co-Authored-By: Claude Sonnet 4.6 --- src/cerbero_mcp/auth.py | 2 +- src/cerbero_mcp/server.py | 205 ++++++++++++++++++++++++++++++++++++++ tests/unit/test_server.py | 62 ++++++++++++ 3 files changed, 268 insertions(+), 1 deletion(-) create mode 100644 src/cerbero_mcp/server.py create mode 100644 tests/unit/test_server.py diff --git a/src/cerbero_mcp/auth.py b/src/cerbero_mcp/auth.py index 08d0a1b..f0feb69 100644 --- a/src/cerbero_mcp/auth.py +++ b/src/cerbero_mcp/auth.py @@ -9,7 +9,7 @@ from fastapi.responses import JSONResponse Environment = Literal["testnet", "mainnet"] -WHITELIST_PATHS = frozenset({"/health", "/apidocs", "/openapi.json"}) +WHITELIST_PATHS = frozenset({"/health", "/apidocs", "/openapi.json", "/docs", "/redoc"}) def _extract_bearer(auth_header: str) -> str | None: diff --git a/src/cerbero_mcp/server.py b/src/cerbero_mcp/server.py new file mode 100644 index 0000000..0471ff8 --- /dev/null +++ b/src/cerbero_mcp/server.py @@ -0,0 +1,205 @@ +"""Factory FastAPI app con middleware, swagger, exception handlers.""" +from __future__ import annotations + +import json +import time +from datetime import UTC, datetime +from typing import Any + +from fastapi import FastAPI, HTTPException, Request +from fastapi.exceptions import RequestValidationError +from fastapi.openapi.utils import get_openapi +from fastapi.responses import JSONResponse, Response +from starlette.middleware.base import BaseHTTPMiddleware + +from cerbero_mcp.auth import install_auth_middleware +from cerbero_mcp.common.errors import ( + HTTP_CODE_MAP, + RETRYABLE_STATUSES, + error_envelope, +) + + +class _TimestampInjectorMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next): + response = await call_next(request) + path = request.url.path + if "/tools/" not in path: + return response + ctype = response.headers.get("content-type", "") + if "application/json" not in ctype: + return response + body = b"" + async for chunk in response.body_iterator: + body += chunk + ts = datetime.now(UTC).isoformat() + try: + data = json.loads(body) if body else None + except Exception: + headers = dict(response.headers) + headers["X-Data-Timestamp"] = ts + return Response( + content=body, status_code=response.status_code, + headers=headers, media_type=response.media_type, + ) + modified = False + if isinstance(data, dict) and "data_timestamp" not in data: + data["data_timestamp"] = ts + modified = True + elif isinstance(data, list): + for item in data: + if isinstance(item, dict) and "data_timestamp" not in item: + item["data_timestamp"] = ts + modified = True + headers = dict(response.headers) + headers["X-Data-Timestamp"] = ts + if modified: + new_body = json.dumps(data, default=str).encode() + headers.pop("content-length", None) + return Response( + content=new_body, status_code=response.status_code, + headers=headers, media_type="application/json", + ) + return Response( + content=body, status_code=response.status_code, + headers=headers, media_type=response.media_type, + ) + + +def build_app( + *, + testnet_token: str, + mainnet_token: str, + title: str = "Cerbero MCP", + version: str = "2.0.0", + description: str = ( + "Multi-exchange MCP server. " + "Bearer token decides environment (testnet/mainnet)." + ), +) -> FastAPI: + app = FastAPI( + title=title, + version=version, + description=description, + docs_url="/apidocs", + redoc_url=None, + openapi_url="/openapi.json", + swagger_ui_parameters={ + "persistAuthorization": True, + "displayRequestDuration": True, + "filter": True, + "tryItOutEnabled": True, + "tagsSorter": "alpha", + "operationsSorter": "alpha", + }, + ) + app.state.boot_at = time.time() + + install_auth_middleware( + app, testnet_token=testnet_token, mainnet_token=mainnet_token + ) + + app.add_middleware(_TimestampInjectorMiddleware) + + @app.middleware("http") + async def latency_header(request: Request, call_next): + t0 = time.perf_counter() + response = await call_next(request) + dur_ms = (time.perf_counter() - t0) * 1000 + response.headers["X-Duration-Ms"] = f"{dur_ms:.2f}" + return response + + @app.exception_handler(HTTPException) + async def _http_exc(request: Request, exc: HTTPException): + retryable = exc.status_code in RETRYABLE_STATUSES + code = HTTP_CODE_MAP.get(exc.status_code, f"HTTP_{exc.status_code}") + message = "HTTP error" + details: dict | None = None + detail = exc.detail + if isinstance(detail, dict): + if isinstance(detail.get("error"), str): + code = detail["error"].upper() + message = str(detail.get("message") or detail.get("error") or message) + details = detail + elif isinstance(detail, str): + message = detail + return JSONResponse( + status_code=exc.status_code, + content=error_envelope( + type_="http_error", code=code, message=message, + retryable=retryable, details=details, + ), + ) + + @app.exception_handler(RequestValidationError) + async def _val_exc(request: Request, exc: RequestValidationError): + errs = exc.errors() + first_loc = ".".join(str(x) for x in errs[0]["loc"]) if errs else "body" + suggestion = ( + f"check field '{first_loc}': " + + (errs[0]["msg"] if errs else "invalid input") + ) + safe_errs: list[dict] = [] + for e in errs[:5]: + ne: dict = {} + for k, v in e.items(): + if k == "ctx" and isinstance(v, dict): + ne[k] = {ck: str(cv) for ck, cv in v.items()} + else: + ne[k] = v + safe_errs.append(ne) + return JSONResponse( + status_code=422, + content=error_envelope( + type_="validation_error", code="INVALID_INPUT", + message=f"request body validation failed on {first_loc}", + retryable=False, suggested_fix=suggestion, + details={"errors": safe_errs}, + ), + ) + + @app.exception_handler(Exception) + async def _unhandled(request: Request, exc: Exception): + return JSONResponse( + status_code=500, + content=error_envelope( + type_="internal_error", code="UNHANDLED_EXCEPTION", + message=f"{type(exc).__name__}: {str(exc)[:300]}", + retryable=True, + ), + ) + + @app.get("/health", tags=["system"]) + def health(): + return { + "status": "healthy", + "name": title, + "version": version, + "uptime_seconds": int(time.time() - app.state.boot_at), + "data_timestamp": datetime.now(UTC).isoformat(), + } + + def _custom_openapi() -> dict[str, Any]: + if app.openapi_schema: + return app.openapi_schema + schema = get_openapi( + title=app.title, version=app.version, + description=app.description, routes=app.routes, + ) + schema.setdefault("components", {}) + schema["components"]["securitySchemes"] = { + "BearerAuth": { + "type": "http", + "scheme": "bearer", + "description": ( + "Use TESTNET_TOKEN for testnet routing, " + "MAINNET_TOKEN for mainnet." + ), + } + } + schema["security"] = [{"BearerAuth": []}] + app.openapi_schema = schema + return schema + + app.openapi = _custom_openapi + return app diff --git a/tests/unit/test_server.py b/tests/unit/test_server.py new file mode 100644 index 0000000..c4f0cac --- /dev/null +++ b/tests/unit/test_server.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import pytest +from fastapi.testclient import TestClient + + +@pytest.fixture +def app(): + from cerbero_mcp.server import build_app + return build_app( + testnet_token="tk_test", + mainnet_token="tk_live", + title="Test", + version="2.0.0", + ) + + +def test_apidocs_served(app): + c = TestClient(app) + r = c.get("/apidocs") + assert r.status_code == 200 + assert "swagger" in r.text.lower() + + +def test_openapi_json_served(app): + c = TestClient(app) + r = c.get("/openapi.json") + assert r.status_code == 200 + spec = r.json() + assert spec["info"]["title"] == "Test" + # securityScheme BearerAuth presente + assert "BearerAuth" in spec["components"]["securitySchemes"] + assert spec["components"]["securitySchemes"]["BearerAuth"]["scheme"] == "bearer" + + +def test_redoc_disabled(app): + c = TestClient(app) + r = c.get("/redoc") + assert r.status_code == 404 + + +def test_default_docs_path_disabled(app): + c = TestClient(app) + r = c.get("/docs") + assert r.status_code == 404 + + +def test_health_endpoint(app): + c = TestClient(app) + r = c.get("/health") + assert r.status_code == 200 + j = r.json() + assert j["status"] == "healthy" + assert j["version"] == "2.0.0" + assert "uptime_seconds" in j + assert "data_timestamp" in j + + +def test_x_duration_ms_header(app): + c = TestClient(app) + r = c.get("/health") + assert "X-Duration-Ms" in r.headers