Coverage for mcpgateway/version.py: 100%

1# -*- coding: utf-8 -*-

2"""Location: ./mcpgateway/version.py

4SPDX-License-Identifier: Apache-2.0

5Authors: Mihai Criveti

7version.py - diagnostics endpoint (HTML + JSON)

8A FastAPI router that mounts at /version and returns either:

9- JSON - machine-readable diagnostics payload

10- HTML - a lightweight dashboard when the client requests text/html or ?format=html

12Features:

13- Cross-platform system metrics (Windows/macOS/Linux), with fallbacks where APIs are unavailable

14- Optional dependencies: psutil (for richer metrics) and redis.asyncio (for Redis health); omitted gracefully if absent

15- Authentication enforcement via `require_auth`; unauthenticated browsers see login form, API clients get JSON 401

16- Redacted environment variables, sanitized DB/Redis URLs

18The module provides comprehensive system diagnostics including application info,

19platform details, database and Redis connectivity, system metrics, and environment

20variables (with secrets redacted).

22Environment variables containing the following patterns are automatically redacted:

23- Keywords: SECRET, TOKEN, PASS, KEY

24- Specific vars: BASIC_AUTH_USER, DATABASE_URL, REDIS_URL

26Examples:

27 >>> from mcpgateway.version import _is_secret, _sanitize_url, START_TIME, HOSTNAME

28 >>> _is_secret("DATABASE_PASSWORD")

29 True

30 >>> _is_secret("BASIC_AUTH_USER")

31 True

32 >>> _is_secret("HOSTNAME")

33 False

34 >>> _sanitize_url("redis://user:xxxxx@localhost:6379/0")

35 'redis://user@localhost:6379/0'

36 >>> _sanitize_url("postgresql://admin:xxxxx@db.example.com/mydb")

37 'postgresql://admin@db.example.com/mydb'

38 >>> _sanitize_url("https://example.com/path")

39 'https://example.com/path'

40 >>> isinstance(START_TIME, float)

41 True

42 >>> START_TIME > 0

43 True

44 >>> isinstance(HOSTNAME, str)

45 True

46 >>> len(HOSTNAME) > 0

47 True

48"""

50# Future

51from __future__ import annotations

53# Standard

54import asyncio

55from datetime import datetime, timezone

56import importlib.util

57import os

58import platform

59import socket

60import time

61from typing import Any, Dict, Optional

62from urllib.parse import urlsplit, urlunsplit

64# Third-Party

65from fastapi import APIRouter, Depends, Request

66from fastapi.responses import HTMLResponse, Response

67from fastapi.templating import Jinja2Templates

68from jinja2 import Environment, FileSystemLoader

69import orjson

70from sqlalchemy import text

72# First-Party

73from mcpgateway import __version__

74from mcpgateway.config import settings

75from mcpgateway.db import engine

76from mcpgateway.utils.orjson_response import ORJSONResponse

77from mcpgateway.utils.redis_client import get_redis_client, is_redis_available

78from mcpgateway.utils.verify_credentials import require_auth

80# Optional runtime dependencies

81try:

82 # Third-Party

83 import psutil # optional for enhanced metrics

84except ImportError:

85 psutil = None # type: ignore

87try:

88 REDIS_AVAILABLE = importlib.util.find_spec("redis.asyncio") is not None

89except (ModuleNotFoundError, AttributeError) as e:

90 # ModuleNotFoundError: redis package not installed

91 # AttributeError: 'redis' exists but isn't a proper package (e.g., shadowed by a file)

92 # Standard

93 import logging

95 logging.getLogger(__name__).warning(f"Redis module check failed ({type(e).__name__}: {e}), Redis support disabled")

96 REDIS_AVAILABLE = False

98# Globals

100START_TIME = time.time()

101HOSTNAME = socket.gethostname()

102LOGIN_PATH = "/login"

103router = APIRouter(tags=["meta"])

104

105

106def _is_secret(key: str) -> bool:

107 """Identify if an environment variable key likely represents a secret.

108

109 Checks if the given environment variable name contains common secret-related

110 keywords or matches specific patterns to prevent accidental exposure of

111 sensitive information in diagnostics.

112

113 Args:

114 key (str): The environment variable name to check.

115

116 Returns:

117 bool: True if the key contains secret-looking keywords or matches

118 known secret patterns, False otherwise.

119

120 Examples:

121 >>> _is_secret("DATABASE_PASSWORD")

122 True

123 >>> _is_secret("API_KEY")

124 True

125 >>> _is_secret("SECRET_TOKEN")

126 True

127 >>> _is_secret("PASS_PHRASE")

128 True

129 >>> # Specific MCP Gateway secrets

130 >>> _is_secret("BASIC_AUTH_USER")

131 True

132 >>> _is_secret("BASIC_AUTH_PASSWORD")

133 True

134 >>> _is_secret("JWT_SECRET_KEY")

135 True

136 >>> _is_secret("AUTH_ENCRYPTION_SECRET")

137 True

138 >>> _is_secret("DATABASE_URL")

139 True

140 >>> _is_secret("REDIS_URL")

141 True

142 >>> # Non-secrets

143 >>> _is_secret("HOSTNAME")

144 False

145 >>> _is_secret("PORT")

146 False

147 >>> _is_secret("DEBUG")

148 False

149 >>> _is_secret("APP_NAME")

150 False

151 >>> # Case insensitive check

152 >>> _is_secret("database_password")

153 True

154 >>> _is_secret("MySecretKey")

155 True

156 >>> _is_secret("basic_auth_user")

157 True

158 >>> _is_secret("redis_url")

159 True

160 """

161 key_upper = key.upper()

162

163 # Check for common secret keywords

164 if any(tok in key_upper for tok in ("SECRET", "TOKEN", "PASS", "KEY")):

165 return True

166

167 # Check for specific secret environment variables

168 secret_vars = {"BASIC_AUTH_USER", "DATABASE_URL", "REDIS_URL"}

169

170 return key_upper in secret_vars

171

172

173def _public_env() -> Dict[str, str]:

174 """Collect environment variables excluding those that look secret.

175

176 Filters out environment variables containing sensitive keywords or matching

177 known secret patterns to create a safe subset for display in diagnostics.

178

179 Returns:

180 Dict[str, str]: A map of environment variable names to values,

181 excluding any variables identified as secrets.

182

183 Examples:

184 >>> import os

185 >>> # Mock environment

186 >>> original_env = dict(os.environ)

187 >>> os.environ.clear()

188 >>> os.environ.update({

189 ... "HOME": "/home/user",

190 ... "PATH": "/usr/bin:/bin",

191 ... "DATABASE_PASSWORD": "xxxxx",

192 ... "API_KEY": "xxxxx",

193 ... "DEBUG": "true",

194 ... "BASIC_AUTH_USER": "admin",

195 ... "BASIC_AUTH_PASSWORD": "xxxxx",

196 ... "JWT_SECRET_KEY": "xxxxx",

197 ... "AUTH_ENCRYPTION_SECRET": "xxxxx",

198 ... "DATABASE_URL": "postgresql://user:xxxxx@localhost/db",

199 ... "REDIS_URL": "redis://user:xxxxx@localhost:6379",

200 ... "APP_NAME": "MyApp",

201 ... "PORT": "8080"

202 ... })

203 >>>

204 >>> result = _public_env()

205 >>> # Public vars should be included

206 >>> "HOME" in result

207 True

208 >>> "PATH" in result

209 True

210 >>> "DEBUG" in result

211 True

212 >>> "APP_NAME" in result

213 True

214 >>> "PORT" in result

215 True

216 >>> # Secrets should be excluded

217 >>> "DATABASE_PASSWORD" in result

218 False

219 >>> "API_KEY" in result

220 False

221 >>> "BASIC_AUTH_USER" in result

222 False

223 >>> "BASIC_AUTH_PASSWORD" in result

224 False

225 >>> "JWT_SECRET_KEY" in result

226 False

227 >>> "AUTH_ENCRYPTION_SECRET" in result

228 False

229 >>> "DATABASE_URL" in result

230 False

231 >>> "REDIS_URL" in result

232 False

233 >>>

234 >>> # Restore original environment

235 >>> os.environ.clear()

236 >>> os.environ.update(original_env)

237 """

238 return {k: v for k, v in os.environ.items() if not _is_secret(k)}

239

240

241def _sanitize_url(url: Optional[str]) -> Optional[str]:

242 """Redact credentials from a URL for safe display.

243

244 Removes password component from URLs while preserving username and other

245 components. Useful for displaying connection strings in logs or diagnostics

246 without exposing sensitive credentials.

247

248 Args:

249 url (Optional[str]): The URL to sanitize, may be None.

250

251 Returns:

252 Optional[str]: The sanitized URL with password removed, or None if input was None.

253

254 Examples:

255 >>> _sanitize_url(None)

256

257 >>> _sanitize_url("")

258

259 >>> # Basic URL without credentials

260 >>> _sanitize_url("http://localhost:8080/path")

261 'http://localhost:8080/path'

262

263 >>> # URL with username and password

264 >>> _sanitize_url("postgresql://user:xxxxx@localhost:5432/db")

265 'postgresql://user@localhost:5432/db'

266

267 >>> # Redis URL with auth

268 >>> _sanitize_url("redis://admin:xxxxx@redis.example.com:6379/0")

269 'redis://admin@redis.example.com:6379/0'

270

271 >>> # URL with only password (no username)

272 >>> _sanitize_url("redis://:xxxxx@localhost:6379")

273 'redis://localhost:6379'

274

275 >>> # Complex URL with query params

276 >>> _sanitize_url("mysql://root:xxxxx@db.local:3306/mydb?charset=utf8")

277 'mysql://root@db.local:3306/mydb?charset=utf8'

278 """

279 if not url:

280 return None

281 parts = urlsplit(url)

282 if parts.password:

283 # Only include username@ if username exists

284 if parts.username:

285 netloc = f"{parts.username}@{parts.hostname}{':' + str(parts.port) if parts.port else ''}"

286 else:

287 netloc = f"{parts.hostname}{':' + str(parts.port) if parts.port else ''}"

288 parts = parts._replace(netloc=netloc)

289 result = urlunsplit(parts)

290 return result if isinstance(result, str) else str(result)

291

292

293def _database_version() -> tuple[str, bool]:

294 """Query the database server version.

295

296 Attempts to connect to the configured database and retrieve its version string.

297 Uses dialect-specific queries for accurate version information.

298

299 Returns:

300 tuple[str, bool]: A tuple containing:

301 - str: Version string on success, or error message on failure

302 - bool: True if database is reachable, False otherwise

303

304 Examples:

305 >>> from unittest.mock import Mock, patch, MagicMock

306 >>>

307 >>> # Test successful SQLite connection

308 >>> mock_engine = Mock()

309 >>> mock_engine.dialect.name = "sqlite"

310 >>> mock_conn = Mock()

311 >>> mock_result = Mock()

312 >>> mock_result.scalar.return_value = "3.39.2"

313 >>> mock_conn.execute.return_value = mock_result

314 >>> mock_conn.__enter__ = Mock(return_value=mock_conn)

315 >>> mock_conn.__exit__ = Mock(return_value=None)

316 >>> mock_engine.connect.return_value = mock_conn

317 >>>

318 >>> with patch('mcpgateway.version.engine', mock_engine):

319 ... version, reachable = _database_version()

320 >>> version

321 '3.39.2'

322 >>> reachable

323 True

324

325 >>> # Test PostgreSQL

326 >>> mock_engine.dialect.name = "postgresql"

327 >>> mock_result.scalar.return_value = "14.5"

328 >>> with patch('mcpgateway.version.engine', mock_engine):

329 ... version, reachable = _database_version()

330 >>> version

331 '14.5'

332 >>> reachable

333 True

334

335 >>> # Test connection failure

336 >>> mock_engine.connect.side_effect = Exception("Connection refused")

337 >>> with patch('mcpgateway.version.engine', mock_engine):

338 ... version, reachable = _database_version()

339 >>> version

340 'Connection refused'

341 >>> reachable

342 False

343 """

344 dialect = engine.dialect.name

345 stmts = {

346 "sqlite": "SELECT sqlite_version();",

347 "postgresql": "SELECT current_setting('server_version');",

348 "mysql": "SELECT version();",

349 }

350 stmt = stmts.get(dialect, "XXSELECT version();XX")

351 try:

352 with engine.connect() as conn:

353 ver = conn.execute(text(stmt)).scalar()

354 return str(ver), True

355 except Exception as exc:

356 return str(exc), False

357

358

359def _system_metrics() -> Dict[str, Any]:

360 """Gather system-wide and per-process metrics using psutil.

361

362 Collects comprehensive system and process metrics with graceful fallbacks

363 when psutil is not installed or certain APIs are unavailable (e.g., on Windows).

364

365 Returns:

366 Dict[str, Any]: A dictionary containing system and process metrics including:

367 - boot_time (str): ISO-formatted system boot time.

368 - cpu_percent (float): Total CPU utilization percentage.

369 - cpu_count (int): Number of logical CPU cores.

370 - cpu_freq_mhz (float | None): Current CPU frequency in MHz (if available).

371 - load_avg (Tuple[float | None, float | None, float | None]): System load average over 1, 5, and 15 minutes,

372 or (None, None, None) if unsupported.

373 - mem_total_mb (float): Total physical memory in MB.

374 - mem_used_mb (float): Used physical memory in MB.

375 - swap_total_mb (float): Total swap memory in MB.

376 - swap_used_mb (float): Used swap memory in MB.

377 - disk_total_gb (float): Total size of the root partition in GB.

378 - disk_used_gb (float): Used space on the root partition in GB.

379 - process (Dict[str, Any]): Dictionary containing metrics for the current process:

380 - pid (int): Current process ID.

381 - threads (int): Number of active threads.

382 - rss_mb (float): Resident Set Size memory usage in MB.

383 - vms_mb (float): Virtual Memory Size usage in MB.

384 - open_fds (int | None): Number of open file descriptors, or None if unsupported.

385 - proc_cpu_percent (float): CPU utilization percentage for the current process.

386

387 Returns empty dict if psutil is not installed.

388

389 Examples:

390 >>> from unittest.mock import Mock, patch

391 >>>

392 >>> # Test without psutil

393 >>> with patch('mcpgateway.version.psutil', None):

394 ... metrics = _system_metrics()

395 >>> metrics

396 {}

397

398 >>> # Test with mocked psutil

399 >>> mock_psutil = Mock()

400 >>> mock_vm = Mock(total=8589934592, used=4294967296) # 8GB total, 4GB used

401 >>> mock_swap = Mock(total=2147483648, used=1073741824) # 2GB total, 1GB used

402 >>> mock_freq = Mock(current=2400.0)

403 >>> mock_disk = Mock(total=107374182400, used=53687091200) # 100GB total, 50GB used

404 >>> mock_mem_info = Mock(rss=104857600, vms=209715200) # 100MB RSS, 200MB VMS

405 >>> mock_process = Mock()

406 >>> mock_process.memory_info.return_value = mock_mem_info

407 >>> mock_process.num_fds.return_value = 42

408 >>> mock_process.cpu_percent.return_value = 25.5

409 >>> mock_process.num_threads.return_value = 4

410 >>> mock_process.pid = 1234

411 >>>

412 >>> mock_psutil.virtual_memory.return_value = mock_vm

413 >>> mock_psutil.swap_memory.return_value = mock_swap

414 >>> mock_psutil.cpu_freq.return_value = mock_freq

415 >>> mock_psutil.cpu_percent.return_value = 45.2

416 >>> mock_psutil.cpu_count.return_value = 8

417 >>> mock_psutil.Process.return_value = mock_process

418 >>> mock_psutil.disk_usage.return_value = mock_disk

419 >>> mock_psutil.boot_time.return_value = 1640995200.0 # 2022-01-01 00:00:00 UTC

420 >>>

421 >>> with patch('mcpgateway.version.psutil', mock_psutil):

422 ... with patch('os.getloadavg', return_value=(1.5, 2.0, 1.75)):

423 ... with patch('os.name', 'posix'):

424 ... metrics = _system_metrics()

425 >>>

426 >>> metrics['cpu_percent']

427 45.2

428 >>> metrics['cpu_count']

429 8

430 >>> metrics['cpu_freq_mhz']

431 2400

432 >>> metrics['load_avg']

433 (1.5, 2.0, 1.75)

434 >>> metrics['mem_total_mb']

435 8192

436 >>> metrics['mem_used_mb']

437 4096

438 >>> metrics['process']['pid']

439 1234

440 >>> metrics['process']['threads']

441 4

442 >>> metrics['process']['rss_mb']

443 100.0

444 >>> metrics['process']['open_fds']

445 42

446 """

447 if not psutil:

448 return {}

449

450 # System memory and swap

451 vm = psutil.virtual_memory()

452 swap = psutil.swap_memory()

453

454 # Load average (Unix); on Windows returns (None, None, None)

455 try:

456 load = tuple(round(x, 2) for x in os.getloadavg())

457 except (AttributeError, OSError):

458 load = (None, None, None)

459

460 # CPU metrics

461 freq = psutil.cpu_freq()

462 cpu_pct = psutil.cpu_percent(interval=0.3)

463 cpu_count = psutil.cpu_count(logical=True)

464

465 # Process metrics

466 proc: "psutil.Process" = psutil.Process()

467 try:

468 open_fds = proc.num_fds()

469 except Exception:

470 open_fds = None

471 proc_cpu_pct = proc.cpu_percent(interval=0.1)

472 memory_info = getattr(proc, "memory_info")()

473 rss_mb = round(memory_info.rss / 1_048_576, 2)

474 vms_mb = round(memory_info.vms / 1_048_576, 2)

475 threads = proc.num_threads()

476 pid = proc.pid

477

478 # Disk usage for root partition (ensure str on Windows)

479 root = os.getenv("SystemDrive", "C:\\") if os.name == "nt" else "/"

480 disk = psutil.disk_usage(str(root))

481 disk_total_gb = round(disk.total / 1_073_741_824, 2)

482 disk_used_gb = round(disk.used / 1_073_741_824, 2)

483

484 return {

485 "boot_time": datetime.fromtimestamp(psutil.boot_time()).isoformat(),

486 "cpu_percent": cpu_pct,

487 "cpu_count": cpu_count,

488 "cpu_freq_mhz": round(freq.current) if freq else None,

489 "load_avg": load,

490 "mem_total_mb": round(vm.total / 1_048_576),

491 "mem_used_mb": round(vm.used / 1_048_576),

492 "swap_total_mb": round(swap.total / 1_048_576),

493 "swap_used_mb": round(swap.used / 1_048_576),

494 "disk_total_gb": disk_total_gb,

495 "disk_used_gb": disk_used_gb,

496 "process": {

497 "pid": pid,

498 "threads": threads,

499 "rss_mb": rss_mb,

500 "vms_mb": vms_mb,

501 "open_fds": open_fds,

502 "proc_cpu_percent": proc_cpu_pct,

503 },

504 }

505

506

507def _build_payload(

508 redis_version: Optional[str],

509 redis_ok: bool,

510) -> Dict[str, Any]:

511 """Build the complete diagnostics payload.

512

513 Assembles all diagnostic information into a structured dictionary suitable

514 for JSON serialization or HTML rendering.

515

516 Args:

517 redis_version (Optional[str]): Redis version string or error message.

518 redis_ok (bool): Whether Redis is reachable and operational.

519

520 Returns:

521 Dict[str, Any]: Complete diagnostics payload containing timestamp, host info,

522 application details, platform info, database and Redis status, settings,

523 environment variables, and system metrics.

524 """

525 db_ver, db_ok = _database_version()

526 return {

527 "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),

528 "host": HOSTNAME,

529 "uptime_seconds": int(time.time() - START_TIME),

530 "app": {

531 "name": settings.app_name,

532 "version": __version__,

533 "mcp_protocol_version": settings.protocol_version,

534 },

535 "platform": {

536 "python": platform.python_version(),

537 "fastapi": __import__("fastapi").__version__,

538 "sqlalchemy": __import__("sqlalchemy").__version__,

539 "os": f"{platform.system()} {platform.release()} ({platform.machine()})",

540 },

541 "database": {

542 "dialect": engine.dialect.name,

543 "url": _sanitize_url(settings.database_url),

544 "reachable": db_ok,

545 "server_version": db_ver,

546 },

547 "redis": {

548 "available": REDIS_AVAILABLE,

549 "url": _sanitize_url(settings.redis_url),

550 "reachable": redis_ok,

551 "server_version": redis_version,

552 },

553 "settings": {

554 "cache_type": settings.cache_type,

555 "mcpgateway_ui_enabled": getattr(settings, "mcpgateway_ui_enabled", None),

556 "mcpgateway_admin_api_enabled": getattr(settings, "mcpgateway_admin_api_enabled", None),

557 "metrics_retention_days": getattr(settings, "metrics_retention_days", 30),

558 "metrics_rollup_retention_days": getattr(settings, "metrics_rollup_retention_days", 365),

559 "metrics_cleanup_enabled": getattr(settings, "metrics_cleanup_enabled", True),

560 "metrics_rollup_enabled": getattr(settings, "metrics_rollup_enabled", True),

561 },

562 "env": _public_env(),

563 "system": _system_metrics(),

564 }

565

566

567def _html_table(obj: Dict[str, Any]) -> str:

568 """Render a dict as an HTML table.

569

570 Converts a dictionary into an HTML table with keys as headers and values

571 as cells. Non-string values are JSON-serialized for display.

572

573 Args:

574 obj (Dict[str, Any]): The dictionary to render as a table.

575

576 Returns:

577 str: HTML table markup string.

578

579 Examples:

580 >>> # Simple string values

581 >>> html = _html_table({"name": "test", "version": "1.0"})

582 >>> '<table>' in html

583 True

584 >>> '<tr><th>name</th><td>test</td></tr>' in html

585 True

586 >>> '<tr><th>version</th><td>1.0</td></tr>' in html

587 True

588

589 >>> # Complex values get JSON serialized

590 >>> html = _html_table({"count": 42, "active": True, "items": ["a", "b"]})

591 >>> '<th>count</th><td>42</td>' in html

592 True

593 >>> '<th>active</th><td>true</td>' in html

594 True

595 >>> '<th>items</th><td>["a","b"]</td>' in html

596 True

597

598 >>> # Empty dict

599 >>> _html_table({})

600 '<table></table>'

601 """

602 rows = "".join(f"<tr><th>{k}</th><td>{orjson.dumps(v, default=str).decode() if not isinstance(v, str) else v}</td></tr>" for k, v in obj.items())

603 return f"<table>{rows}</table>"

604

605

606def _render_html(payload: Dict[str, Any]) -> str:

607 """Render the full diagnostics payload as HTML.

608

609 Creates a complete HTML page with styled tables displaying all diagnostic

610 information in a user-friendly format.

611

612 Args:

613 payload (Dict[str, Any]): The complete diagnostics data structure.

614

615 Returns:

616 str: Complete HTML page as a string.

617

618 Examples:

619 >>> payload = {

620 ... "timestamp": "2024-01-01T00:00:00Z",

621 ... "host": "test-server",

622 ... "uptime_seconds": 3600,

623 ... "app": {"name": "TestApp", "version": "1.0"},

624 ... "platform": {"python": "3.9.0"},

625 ... "database": {"dialect": "sqlite", "reachable": True},

626 ... "redis": {"available": False},

627 ... "settings": {"cache_type": "memory"},

628 ... "system": {"cpu_count": 4},

629 ... "env": {"PATH": "/usr/bin"}

630 ... }

631 >>>

632 >>> html = _render_html(payload)

633 >>> '<!doctype html>' in html

634 True

635 >>> '<h1>MCP Gateway diagnostics</h1>' in html

636 True

637 >>> 'test-server' in html

638 True

639 >>> '3600s' in html

640 True

641 >>> '<h2>App</h2>' in html

642 True

643 >>> '<h2>Database</h2>' in html

644 True

645 >>> '<style>' in html

646 True

647 >>> 'border-collapse:collapse' in html

648 True

649 """

650 style = (

651 "<style>"

652 "body{font-family:system-ui,sans-serif;margin:2rem;}"

653 "table{border-collapse:collapse;width:100%;margin-bottom:1rem;}"

654 "th,td{border:1px solid #ccc;padding:.5rem;text-align:left;}"

655 "th{background:#f7f7f7;width:25%;}"

656 "</style>"

657 )

658 header = f"<h1>MCP Gateway diagnostics</h1><p>Generated {payload['timestamp']} - Host {payload['host']} - Uptime {payload['uptime_seconds']}s</p>"

659 sections = ""

660 for title, key in (

661 ("App", "app"),

662 ("Platform", "platform"),

663 ("Database", "database"),

664 ("Redis", "redis"),

665 ("Settings", "settings"),

666 ("System", "system"),

667 ):

668 sections += f"<h2>{title}</h2>{_html_table(payload[key])}"

669 env_section = f"<h2>Environment</h2>{_html_table(payload['env'])}"

670 return f"<!doctype html><html><head><meta charset='utf-8'>{style}</head><body>{header}{sections}{env_section}</body></html>"

671

672

673def _login_html(next_url: str) -> str:

674 """Render the login form HTML for unauthenticated browsers.

675

676 Creates a simple login form that posts credentials and redirects back

677 to the requested URL after successful authentication.

678

679 Args:

680 next_url (str): The URL to redirect to after successful login.

681

682 Returns:

683 str: HTML string containing the complete login page.

684

685 Examples:

686 >>> html = _login_html("/version?format=html")

687 >>> '<!doctype html>' in html

688 True

689 >>> '<h2>Please log in</h2>' in html

690 True

691 >>> 'action="/login"' in html

692 True

693 >>> 'name="next" value="/version?format=html"' in html

694 True

695 >>> 'type="text" name="username"' in html

696 True

697 >>> 'type="password" name="password"' in html

698 True

699 >>> 'autocomplete="username"' in html

700 True

701 >>> 'autocomplete="current-password"' in html

702 True

703 >>> '<button type="submit">Login</button>' in html

704 True

705 """

706 return f"""<!doctype html>

707<html><head><meta charset='utf-8'><title>Login - MCP Gateway</title>

708<style>

709body{{font-family:system-ui,sans-serif;margin:2rem;}}

710form{{max-width:320px;margin:auto;}}

711label{{display:block;margin:.5rem 0;}}

712input{{width:100%;padding:.5rem;}}

713button{{margin-top:1rem;padding:.5rem 1rem;}}

714</style></head>

715<body>

716 <h2>Please log in</h2>

717 <form action="{LOGIN_PATH}" method="post">

718 <input type="hidden" name="next" value="{next_url}">

719 <label>Username<input type="text" name="username" autocomplete="username"></label>

720 <label>Password<input type="password" name="password" autocomplete="current-password"></label>

721 <button type="submit">Login</button>

722 </form>

723</body></html>"""

724

725

726# Endpoint

727@router.get("/version", summary="Diagnostics (auth required)")

728async def version_endpoint(

729 request: Request,

730 fmt: Optional[str] = None,

731 partial: Optional[bool] = False,

732 _user=Depends(require_auth),

733) -> Response:

734 """Serve diagnostics as JSON, full HTML, or partial HTML.

735

736 Main endpoint that gathers all diagnostic information and returns it in the

737 requested format. Requires authentication via HTTP Basic Auth or session.

738

739 The endpoint supports three output formats:

740 - JSON (default): Machine-readable diagnostic data

741 - Full HTML: Complete HTML page with styled tables

742 - Partial HTML: HTML fragment for embedding (when partial=True)

743

744 Args:

745 request (Request): The incoming FastAPI request object.

746 fmt (Optional[str]): Query parameter to force format ('html' for HTML output).

747 partial (Optional[bool]): Query parameter to request partial HTML fragment.

748 _user: Injected authenticated user from require_auth dependency.

749

750 Returns:

751 Response: JSONResponse with diagnostic data, or HTMLResponse with formatted page.

752

753 Examples:

754 >>> import asyncio

755 >>> from unittest.mock import Mock, AsyncMock, patch

756 >>> from fastapi import Request

757 >>> from fastapi.responses import JSONResponse, HTMLResponse

758 >>>

759 >>> # Create mock request

760 >>> mock_request = Mock(spec=Request)

761 >>> mock_request.headers = {"accept": "application/json"}

762 >>>

763 >>> # Test JSON response (default)

764 >>> async def test_json():

765 ... with patch('mcpgateway.version.REDIS_AVAILABLE', False):

766 ... with patch('mcpgateway.version._build_payload') as mock_build:

767 ... mock_build.return_value = {"test": "data"}

768 ... response = await version_endpoint(mock_request, fmt=None, partial=False, _user="testuser")

769 ... return response

770 >>>

771 >>> response = asyncio.run(test_json())

772 >>> isinstance(response, JSONResponse)

773 True

774

775 >>> # Test HTML response with fmt parameter

776 >>> async def test_html_fmt():

777 ... with patch('mcpgateway.version.REDIS_AVAILABLE', False):

778 ... with patch('mcpgateway.version._build_payload') as mock_build:

779 ... with patch('mcpgateway.version._render_html') as mock_render:

780 ... mock_build.return_value = {"test": "data"}

781 ... mock_render.return_value = "<html>test</html>"

782 ... response = await version_endpoint(mock_request, fmt="html", partial=False, _user="testuser")

783 ... return response

784 >>>

785 >>> response = asyncio.run(test_html_fmt())

786 >>> isinstance(response, HTMLResponse)

787 True

788

789 >>> # Test with Redis available (using is_redis_available and get_redis_client)

790 >>> async def test_with_redis():

791 ... from mcpgateway.utils.redis_client import _reset_client

792 ... _reset_client() # Reset shared client state for clean test

793 ... mock_redis = AsyncMock()

794 ... mock_redis.info = AsyncMock(return_value={"redis_version": "7.0.5"})

795 ...

796 ... async def mock_get_redis_client():

797 ... return mock_redis

798 ...

799 ... async def mock_is_redis_available():

800 ... return True

801 ...

802 ... with patch('mcpgateway.version.REDIS_AVAILABLE', True):

803 ... with patch('mcpgateway.version.settings') as mock_settings:

804 ... mock_settings.cache_type = "redis"

805 ... mock_settings.redis_url = "redis://localhost:6379"

806 ... with patch('mcpgateway.version.is_redis_available', mock_is_redis_available):

807 ... with patch('mcpgateway.version.get_redis_client', mock_get_redis_client):

808 ... with patch('mcpgateway.version._build_payload') as mock_build:

809 ... mock_build.return_value = {"redis": {"version": "7.0.5"}}

810 ... response = await version_endpoint(mock_request, _user="testuser")

811 ... # Verify Redis info was retrieved

812 ... mock_redis.info.assert_called_once()

813 ... # Verify payload was built with Redis info

814 ... mock_build.assert_called_once_with("7.0.5", True)

815 ... _reset_client() # Clean up after test

816 ... return response

817 >>>

818 >>> response = asyncio.run(test_with_redis())

819 >>> isinstance(response, JSONResponse)

820 True

821 """

822 # Redis health check - use shared client from factory

823 redis_ok = False

824 redis_version: Optional[str] = None

825 if REDIS_AVAILABLE and settings.cache_type.lower() == "redis" and settings.redis_url:

826 try:

827 # Use centralized availability check

828 redis_ok = await is_redis_available()

829 if redis_ok:

830 client = await get_redis_client()

831 if client:

832 info = await asyncio.wait_for(client.info(), timeout=3.0)

833 redis_version = info.get("redis_version", "unknown")

834 else:

835 redis_version = "Client not available"

836 else:

837 redis_version = "Not reachable"

838 except Exception as exc:

839 redis_ok = False

840 redis_version = str(exc)

841

842 payload = _build_payload(redis_version, redis_ok)

843 if partial:

844 # Return partial HTML fragment for HTMX embedding

845 templates = getattr(request.app.state, "templates", None)

846 if templates is None:

847 jinja_env = Environment(

848 loader=FileSystemLoader(str(settings.templates_dir)),

849 autoescape=True,

850 auto_reload=settings.templates_auto_reload,

851 )

852 templates = Jinja2Templates(env=jinja_env)

853 return templates.TemplateResponse(request, "version_info_partial.html", {"request": request, "payload": payload})

854 wants_html = fmt == "html" or "text/html" in request.headers.get("accept", "")

855 if wants_html:

856 return HTMLResponse(_render_html(payload))

857 return ORJSONResponse(payload)

Coverage for mcpgateway / version.py: 100%

137 statements