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_admin_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 ContextForge 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

173_PUBLIC_ENV_PREFIXES = ("MCPGATEWAY_", "MCP_")

174_PUBLIC_ENV_ALLOWLIST = frozenset(

175 {

176 "PORT",

177 "HOST",

178 "RELOAD",

179 "LOG_LEVEL",

180 "LOG_TO_FILE",

181 "PLUGINS_ENABLED",

182 "OBSERVABILITY_ENABLED",

183 "AUTH_REQUIRED",

184 "ALLOWED_ORIGINS",

185 }

186)

187

188

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

190 """Collect application-specific environment variables for diagnostics.

191

192 Only returns variables with ``MCPGATEWAY_`` or ``MCP_`` prefixes, plus a

193 curated allowlist of safe operational variables. Secrets are still excluded

194 via :func:`_is_secret`.

195

196 Returns:

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

198

199 Examples:

200 >>> import os

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

202 >>> os.environ.clear()

203 >>> os.environ.update({

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

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

206 ... "PORT": "8080",

207 ... "HOST": "0.0.0.0",

208 ... "MCPGATEWAY_UI_ENABLED": "true",

209 ... "MCP_REQUIRE_AUTH": "true",

210 ... "DATABASE_PASSWORD": "xxxxx",

211 ... "JWT_SECRET_KEY": "xxxxx",

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

213 ... })

214 >>>

215 >>> result = _public_env()

216 >>> # App-prefixed vars included

217 >>> "MCPGATEWAY_UI_ENABLED" in result

218 True

219 >>> "MCP_REQUIRE_AUTH" in result

220 True

221 >>> # Allowlisted vars included

222 >>> "PORT" in result

223 True

224 >>> "HOST" in result

225 True

226 >>> # System vars excluded

227 >>> "HOME" in result

228 False

229 >>> "PATH" in result

230 False

231 >>> # Secrets still excluded

232 >>> "DATABASE_PASSWORD" in result

233 False

234 >>> "JWT_SECRET_KEY" in result

235 False

236 >>> "DATABASE_URL" in result

237 False

238 >>>

239 >>> os.environ.clear()

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

241 """

242 return {k: v for k, v in os.environ.items() if not _is_secret(k) and (k.upper().startswith(_PUBLIC_ENV_PREFIXES) or k.upper() in _PUBLIC_ENV_ALLOWLIST)}

243

244

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

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

247

248 Removes password component from URLs while preserving username and other

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

250 without exposing sensitive credentials.

251

252 Args:

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

254

255 Returns:

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

257

258 Examples:

259 >>> _sanitize_url(None)

260

261 >>> _sanitize_url("")

262

263 >>> # Basic URL without credentials

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

265 'http://localhost:8080/path'

266

267 >>> # URL with username and password

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

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

270

271 >>> # Redis URL with auth

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

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

274

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

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

277 'redis://localhost:6379'

278

279 >>> # Complex URL with query params

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

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

282 """

283 if not url:

284 return None

285 parts = urlsplit(url)

286 if parts.password:

287 # Only include username@ if username exists

288 if parts.username:

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

290 else:

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

292 parts = parts._replace(netloc=netloc)

293 result = urlunsplit(parts)

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

295

296

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

298 """Query the database server version.

299

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

301 Uses dialect-specific queries for accurate version information.

302

303 Returns:

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

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

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

307

308 Examples:

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

310 >>>

311 >>> # Test successful SQLite connection

312 >>> mock_engine = Mock()

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

314 >>> mock_conn = Mock()

315 >>> mock_result = Mock()

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

317 >>> mock_conn.execute.return_value = mock_result

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

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

320 >>> mock_engine.connect.return_value = mock_conn

321 >>>

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

323 ... version, reachable = _database_version()

324 >>> version

325 '3.39.2'

326 >>> reachable

327 True

328

329 >>> # Test PostgreSQL

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

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

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

333 ... version, reachable = _database_version()

334 >>> version

335 '14.5'

336 >>> reachable

337 True

338

339 >>> # Test connection failure

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

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

342 ... version, reachable = _database_version()

343 >>> version

344 'Connection refused'

345 >>> reachable

346 False

347 """

348 dialect = engine.dialect.name

349 stmts = {

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

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

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

353 }

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

355 try:

356 with engine.connect() as conn:

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

358 return str(ver), True

359 except Exception as exc:

360 return str(exc), False

361

362

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

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

365

366 Collects comprehensive system and process metrics with graceful fallbacks

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

368

369 Returns:

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

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

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

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

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

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

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

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

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

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

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

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

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

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

384 - pid (int): Current process ID.

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

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

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

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

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

390

391 Returns empty dict if psutil is not installed.

392

393 Examples:

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

395 >>>

396 >>> # Test without psutil

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

398 ... metrics = _system_metrics()

399 >>> metrics

400 {}

401

402 >>> # Test with mocked psutil

403 >>> mock_psutil = Mock()

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

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

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

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

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

409 >>> mock_process = Mock()

410 >>> mock_process.memory_info.return_value = mock_mem_info

411 >>> mock_process.num_fds.return_value = 42

412 >>> mock_process.cpu_percent.return_value = 25.5

413 >>> mock_process.num_threads.return_value = 4

414 >>> mock_process.pid = 1234

415 >>>

416 >>> mock_psutil.virtual_memory.return_value = mock_vm

417 >>> mock_psutil.swap_memory.return_value = mock_swap

418 >>> mock_psutil.cpu_freq.return_value = mock_freq

419 >>> mock_psutil.cpu_percent.return_value = 45.2

420 >>> mock_psutil.cpu_count.return_value = 8

421 >>> mock_psutil.Process.return_value = mock_process

422 >>> mock_psutil.disk_usage.return_value = mock_disk

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

424 >>>

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

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

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

428 ... metrics = _system_metrics()

429 >>>

430 >>> metrics['cpu_percent']

431 45.2

432 >>> metrics['cpu_count']

433 8

434 >>> metrics['cpu_freq_mhz']

435 2400

436 >>> metrics['load_avg']

437 (1.5, 2.0, 1.75)

438 >>> metrics['mem_total_mb']

439 8192

440 >>> metrics['mem_used_mb']

441 4096

442 >>> metrics['process']['pid']

443 1234

444 >>> metrics['process']['threads']

445 4

446 >>> metrics['process']['rss_mb']

447 100.0

448 >>> metrics['process']['open_fds']

449 42

450 """

451 if not psutil:

452 return {}

453

454 # System memory and swap

455 vm = psutil.virtual_memory()

456 swap = psutil.swap_memory()

457

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

459 try:

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

461 except (AttributeError, OSError):

462 load = (None, None, None)

463

464 # CPU metrics

465 freq = psutil.cpu_freq()

466 cpu_pct = psutil.cpu_percent(interval=0.3)

467 cpu_count = psutil.cpu_count(logical=True)

468

469 # Process metrics

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

471 try:

472 open_fds = proc.num_fds()

473 except Exception:

474 open_fds = None

475 proc_cpu_pct = proc.cpu_percent(interval=0.1)

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

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

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

479 threads = proc.num_threads()

480 pid = proc.pid

481

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

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

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

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

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

487

488 return {

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

490 "cpu_percent": cpu_pct,

491 "cpu_count": cpu_count,

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

493 "load_avg": load,

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

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

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

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

498 "disk_total_gb": disk_total_gb,

499 "disk_used_gb": disk_used_gb,

500 "process": {

501 "pid": pid,

502 "threads": threads,

503 "rss_mb": rss_mb,

504 "vms_mb": vms_mb,

505 "open_fds": open_fds,

506 "proc_cpu_percent": proc_cpu_pct,

507 },

508 }

509

510

511def _build_payload(

512 redis_version: Optional[str],

513 redis_ok: bool,

514) -> Dict[str, Any]:

515 """Build the complete diagnostics payload.

516

517 Assembles all diagnostic information into a structured dictionary suitable

518 for JSON serialization or HTML rendering.

519

520 Args:

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

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

523

524 Returns:

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

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

527 environment variables, and system metrics.

528 """

529 db_ver, db_ok = _database_version()

530 return {

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

532 "host": HOSTNAME,

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

534 "app": {

535 "name": settings.app_name,

536 "version": __version__,

537 "mcp_protocol_version": settings.protocol_version,

538 },

539 "platform": {

540 "python": platform.python_version(),

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

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

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

544 },

545 "database": {

546 "dialect": engine.dialect.name,

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

548 "reachable": db_ok,

549 "server_version": db_ver,

550 },

551 "redis": {

552 "available": REDIS_AVAILABLE,

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

554 "reachable": redis_ok,

555 "server_version": redis_version,

556 },

557 "settings": {

558 "cache_type": settings.cache_type,

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

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

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

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

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

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

565 },

566 "env": _public_env(),

567 "system": _system_metrics(),

568 }

569

570

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

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

573

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

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

576

577 Args:

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

579

580 Returns:

581 str: HTML table markup string.

582

583 Examples:

584 >>> # Simple string values

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

586 >>> '<table>' in html

587 True

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

589 True

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

591 True

592

593 >>> # Complex values get JSON serialized

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

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

596 True

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

598 True

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

600 True

601

602 >>> # Empty dict

603 >>> _html_table({})

604 '<table></table>'

605 """

606 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())

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

608

609

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

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

612

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

614 information in a user-friendly format.

615

616 Args:

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

618

619 Returns:

620 str: Complete HTML page as a string.

621

622 Examples:

623 >>> payload = {

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

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

626 ... "uptime_seconds": 3600,

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

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

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

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

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

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

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

634 ... }

635 >>>

636 >>> html = _render_html(payload)

637 >>> '<!doctype html>' in html

638 True

639 >>> '<h1>ContextForge diagnostics</h1>' in html

640 True

641 >>> 'test-server' in html

642 True

643 >>> '3600s' in html

644 True

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

646 True

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

648 True

649 >>> '<style>' in html

650 True

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

652 True

653 """

654 style = (

655 "<style>"

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

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

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

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

660 "</style>"

661 )

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

663 sections = ""

664 for title, key in (

665 ("App", "app"),

666 ("Platform", "platform"),

667 ("Database", "database"),

668 ("Redis", "redis"),

669 ("Settings", "settings"),

670 ("System", "system"),

671 ):

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

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

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

675

676

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

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

679

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

681 to the requested URL after successful authentication.

682

683 Args:

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

685

686 Returns:

687 str: HTML string containing the complete login page.

688

689 Examples:

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

691 >>> '<!doctype html>' in html

692 True

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

694 True

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

696 True

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

698 True

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

700 True

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

702 True

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

704 True

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

706 True

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

708 True

709 """

710 return f"""<!doctype html>

711<html><head><meta charset='utf-8'><title>Login - ContextForge</title>

712<style>

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

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

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

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

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

718</style></head>

719<body>

720 <h2>Please log in</h2>

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

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

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

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

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

726 </form>

727</body></html>"""

728

729

730# Endpoint

731@router.get("/version", summary="Diagnostics (admin only)")

732async def version_endpoint(

733 request: Request,

734 fmt: Optional[str] = None,

735 partial: Optional[bool] = False,

736 _user=Depends(require_admin_auth),

737) -> Response:

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

739

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

741 requested format. Requires admin authentication.

742

743 The endpoint supports three output formats:

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

745 - Full HTML: Complete HTML page with styled tables

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

747

748 Args:

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

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

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

752 _user: Injected authenticated admin user from require_admin_auth dependency.

753

754 Returns:

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

756

757 Examples:

758 >>> import asyncio

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

760 >>> from fastapi import Request

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

762 >>>

763 >>> # Create mock request

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

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

766 >>>

767 >>> # Test JSON response (default)

768 >>> async def test_json():

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

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

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

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

773 ... return response

774 >>>

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

776 >>> isinstance(response, JSONResponse)

777 True

778

779 >>> # Test HTML response with fmt parameter

780 >>> async def test_html_fmt():

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

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

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

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

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

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

787 ... return response

788 >>>

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

790 >>> isinstance(response, HTMLResponse)

791 True

792

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

794 >>> async def test_with_redis():

795 ... from mcpgateway.utils.redis_client import _reset_client

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

797 ... mock_redis = AsyncMock()

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

799 ...

800 ... async def mock_get_redis_client():

801 ... return mock_redis

802 ...

803 ... async def mock_is_redis_available():

804 ... return True

805 ...

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

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

808 ... mock_settings.cache_type = "redis"

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

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

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

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

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

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

815 ... # Verify Redis info was retrieved

816 ... mock_redis.info.assert_called_once()

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

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

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

820 ... return response

821 >>>

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

823 >>> isinstance(response, JSONResponse)

824 True

825 """

826 # Redis health check - use shared client from factory

827 redis_ok = False

828 redis_version: Optional[str] = None

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

830 try:

831 # Use centralized availability check

832 redis_ok = await is_redis_available()

833 if redis_ok:

834 client = await get_redis_client()

835 if client:

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

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

838 else:

839 redis_version = "Client not available"

840 else:

841 redis_version = "Not reachable"

842 except Exception as exc:

843 redis_ok = False

844 redis_version = str(exc)

845

846 payload = _build_payload(redis_version, redis_ok)

847 if partial:

848 # Return partial HTML fragment for HTMX embedding

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

850 if templates is None:

851 jinja_env = Environment(

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

853 autoescape=True,

854 auto_reload=settings.templates_auto_reload,

855 )

856 templates = Jinja2Templates(env=jinja_env)

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

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

859 if wants_html:

860 return HTMLResponse(_render_html(payload))

861 return ORJSONResponse(payload)

Coverage for mcpgateway / version.py: 100%

139 statements