Lots of cleanup of secrets in code

Signed-off-by: Jonathan Springer <[email protected]>

Author: jonpspri

mcpgateway/config.py

@@ -49,7 +49,6 @@
 
 # Standard
 from functools import lru_cache
-from importlib.abc import Traversable
 from importlib.resources import files
 import json  # TODO: consider typjson for type safety loading from configuration data.
 import logging
@@ -155,8 +154,9 @@ class Settings(BaseSettings):
     database_url: str = "sqlite:///./mcp.db"
 
     # Absolute paths resolved at import-time (still override-able via env vars)
-    templates_dir: Traversable = files("mcpgateway") / "templates"
-    static_dir: Traversable = files("mcpgateway") / "static"
+    templates_dir: Path = Field(default_factory=lambda: Path(str(files("mcpgateway") / "templates")))
+    static_dir: Path = Field(default_factory=lambda: Path(str(files("mcpgateway") / "static")))
+
     app_root_path: str = ""
 
     # Protocol

mcpgateway/main.py

@@ -104,9 +104,8 @@
 from mcpgateway.services.completion_service import CompletionService
 from mcpgateway.services.export_service import ExportError, ExportService
 from mcpgateway.services.gateway_service import GatewayConnectionError, GatewayError, GatewayNameConflictError, GatewayNotFoundError, GatewayService, GatewayUrlConflictError
-from mcpgateway.services.import_service import ConflictStrategy, ImportConflictError
+from mcpgateway.services.import_service import ConflictStrategy, ImportConflictError, ImportService, ImportValidationError
 from mcpgateway.services.import_service import ImportError as ImportServiceError
-from mcpgateway.services.import_service import ImportService, ImportValidationError
 from mcpgateway.services.logging_service import LoggingService
 from mcpgateway.services.prompt_service import PromptError, PromptNameConflictError, PromptNotFoundError, PromptService
 from mcpgateway.services.resource_service import ResourceError, ResourceNotFoundError, ResourceService, ResourceURIConflictError
@@ -1082,9 +1081,10 @@ def require_api_key(api_key: str) -> None:
 
     Examples:
         >>> from mcpgateway.config import settings
+        >>> from pydantic import SecretStr
         >>> settings.auth_required = True
         >>> settings.basic_auth_user = "admin"
-        >>> settings.basic_auth_password = "secret"
+        >>> settings.basic_auth_password = SecretStr("secret")
         >>>
         >>> # Valid API key
         >>> require_api_key("admin:secret")  # Should not raise
@@ -2694,8 +2694,10 @@ async def read_resource(resource_id: str, request: Request, db: Session = Depend
     # Ensure a plain JSON-serializable structure
     try:
         # First-Party
-        from mcpgateway.models import ResourceContent  # pylint: disable=import-outside-toplevel
-        from mcpgateway.models import TextContent  # pylint: disable=import-outside-toplevel
+        from mcpgateway.models import (
+            ResourceContent,  # pylint: disable=import-outside-toplevel
+            TextContent,  # pylint: disable=import-outside-toplevel
+        )
 
         # If already a ResourceContent, serialize directly
         if isinstance(content, ResourceContent):

mcpgateway/scripts/validate_env.py

@@ -53,7 +53,8 @@ def get_security_warnings(settings: Settings) -> list[str]:
         warnings.append(f"PORT: Out of allowed range (1-65535). Got: {settings.port}")
 
     # --- PLATFORM_ADMIN_PASSWORD ---
-    pw = settings.platform_admin_password
+    pw = settings.platform_admin_password.get_secret_value() if isinstance(settings.platform_admin_password, SecretStr) else settings.platform_admin_password
+
     if not pw or pw.lower() in ("changeme", "admin", "password"):
         warnings.append("Default admin password detected! Please change PLATFORM_ADMIN_PASSWORD immediately.")
     min_length = settings.password_min_length
@@ -64,7 +65,7 @@ def get_security_warnings(settings: Settings) -> list[str]:
         warnings.append("Admin password has low complexity. Should contain at least 3 of: uppercase, lowercase, digits, special characters")
 
     # --- BASIC_AUTH_PASSWORD ---
-    basic_pw = settings.basic_auth_password
+    basic_pw = settings.basic_auth_password.get_secret_value() if isinstance(settings.basic_auth_password, SecretStr) else settings.basic_auth_password
     if not basic_pw or basic_pw.lower() in ("changeme", "password"):
         warnings.append("Default BASIC_AUTH_PASSWORD detected! Please change it immediately.")
     min_length = settings.password_min_length

mcpgateway/utils/verify_credentials.py

@@ -10,14 +10,15 @@
 headers and cookies.
 Examples:
     >>> from mcpgateway.utils import verify_credentials as vc
+    >>> from pydantic import SecretStr
     >>> class DummySettings:
     ...     jwt_secret_key = 'secret'
     ...     jwt_algorithm = 'HS256'
     ...     jwt_audience = 'mcpgateway-api'
     ...     jwt_issuer = 'mcpgateway'
     ...     jwt_audience_verification = True
     ...     basic_auth_user = 'user'
-    ...     basic_auth_password = 'pass'
+    ...     basic_auth_password = SecretStr('pass')
     ...     auth_required = True
     ...     require_token_expiration = False
     ...     docs_allow_basic_auth = False
@@ -153,14 +154,15 @@ async def verify_credentials(token: str) -> dict:
 
     Examples:
         >>> from mcpgateway.utils import verify_credentials as vc
+        >>> from pydantic import SecretStr
         >>> class DummySettings:
         ...     jwt_secret_key = 'secret'
         ...     jwt_algorithm = 'HS256'
         ...     jwt_audience = 'mcpgateway-api'
         ...     jwt_issuer = 'mcpgateway'
         ...     jwt_audience_verification = True
         ...     basic_auth_user = 'user'
-        ...     basic_auth_password = 'pass'
+        ...     basic_auth_password = SecretStr('pass')
         ...     auth_required = True
         ...     require_token_expiration = False
         ...     docs_allow_basic_auth = False
@@ -202,14 +204,15 @@ async def require_auth(request: Request, credentials: Optional[HTTPAuthorization
 
     Examples:
         >>> from mcpgateway.utils import verify_credentials as vc
+        >>> from pydantic import SecretStr
         >>> class DummySettings:
         ...     jwt_secret_key = 'secret'
         ...     jwt_algorithm = 'HS256'
         ...     jwt_audience = 'mcpgateway-api'
         ...     jwt_issuer = 'mcpgateway'
         ...     jwt_audience_verification = True
         ...     basic_auth_user = 'user'
-        ...     basic_auth_password = 'pass'
+        ...     basic_auth_password = SecretStr('pass')
         ...     auth_required = True
         ...     mcp_client_auth_enabled = True
         ...     trust_proxy_auth = False
@@ -306,14 +309,15 @@ async def verify_basic_credentials(credentials: HTTPBasicCredentials) -> str:
 
     Examples:
         >>> from mcpgateway.utils import verify_credentials as vc
+        >>> from pydantic import SecretStr
         >>> class DummySettings:
         ...     jwt_secret_key = 'secret'
         ...     jwt_algorithm = 'HS256'
         ...     jwt_audience = 'mcpgateway-api'
         ...     jwt_issuer = 'mcpgateway'
         ...     jwt_audience_verification = True
         ...     basic_auth_user = 'user'
-        ...     basic_auth_password = 'pass'
+        ...     basic_auth_password = SecretStr('pass')
         ...     auth_required = True
         ...     docs_allow_basic_auth = False
         >>> vc.settings = DummySettings()
@@ -330,7 +334,7 @@ async def verify_basic_credentials(credentials: HTTPBasicCredentials) -> str:
         error
     """
     is_valid_user = credentials.username == settings.basic_auth_user
-    is_valid_pass = credentials.password == settings.basic_auth_password
+    is_valid_pass = credentials.password == settings.basic_auth_password.get_secret_value()
 
     if not (is_valid_user and is_valid_pass):
         raise HTTPException(
@@ -359,14 +363,15 @@ async def require_basic_auth(credentials: HTTPBasicCredentials = Depends(basic_s
 
     Examples:
         >>> from mcpgateway.utils import verify_credentials as vc
+        >>> from pydantic import SecretStr
         >>> class DummySettings:
         ...     jwt_secret_key = 'secret'
         ...     jwt_algorithm = 'HS256'
         ...     jwt_audience = 'mcpgateway-api'
         ...     jwt_issuer = 'mcpgateway'
         ...     jwt_audience_verification = True
         ...     basic_auth_user = 'user'
-        ...     basic_auth_password = 'pass'
+        ...     basic_auth_password = SecretStr('pass')
         ...     auth_required = True
         ...     docs_allow_basic_auth = False
         >>> vc.settings = DummySettings()
@@ -420,14 +425,15 @@ async def require_docs_basic_auth(auth_header: str) -> str:
 
     Examples:
         >>> from mcpgateway.utils import verify_credentials as vc
+        >>> from pydantic import SecretStr
         >>> class DummySettings:
         ...     jwt_secret_key = 'secret'
         ...     jwt_algorithm = 'HS256'
         ...     jwt_audience = 'mcpgateway-api'
         ...     jwt_issuer = 'mcpgateway'
         ...     jwt_audience_verification = True
         ...     basic_auth_user = 'user'
-        ...     basic_auth_password = 'pass'
+        ...     basic_auth_password = SecretStr('pass')
         ...     auth_required = True
         ...     require_token_expiration = False
         ...     docs_allow_basic_auth = True
@@ -633,14 +639,15 @@ async def require_auth_override(
 
     Examples:
         >>> from mcpgateway.utils import verify_credentials as vc
+        >>> from pydantic import SecretStr
         >>> class DummySettings:
         ...     jwt_secret_key = 'secret'
         ...     jwt_algorithm = 'HS256'
         ...     jwt_audience = 'mcpgateway-api'
         ...     jwt_issuer = 'mcpgateway'
         ...     jwt_audience_verification = True
         ...     basic_auth_user = 'user'
-        ...     basic_auth_password = 'pass'
+        ...     basic_auth_password = SecretStr('pass')
         ...     auth_required = True
         ...     mcp_client_auth_enabled = True
         ...     trust_proxy_auth = False

tests/unit/mcpgateway/test_bootstrap_db.py

@@ -12,6 +12,7 @@
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 # Third-Party
+from pydantic import SecretStr
 import pytest
 
 # First-Party
@@ -30,7 +31,7 @@ def mock_settings():
     settings = Mock()
     settings.email_auth_enabled = True
     settings.platform_admin_email = "[email protected]"
-    settings.platform_admin_password = "secure_password"
+    settings.platform_admin_password = SecretStr("secure_password")
     settings.platform_admin_full_name = "Platform Admin"
     settings.auto_create_personal_teams = True
     settings.database_url = "sqlite:///:memory:"
@@ -124,20 +125,25 @@ async def test_bootstrap_admin_user_success(self, mock_settings, mock_db_session
         mock_email_auth_service.get_user_by_email.return_value = None
         mock_email_auth_service.create_user.return_value = mock_admin_user
 
-        with patch("mcpgateway.bootstrap_db.settings", mock_settings):
-            with patch("mcpgateway.bootstrap_db.SessionLocal", return_value=mock_db_session):
-                with patch("mcpgateway.services.email_auth_service.EmailAuthService", return_value=mock_email_auth_service):
-                    with patch("mcpgateway.db.utc_now") as mock_utc_now:
-                        mock_utc_now.return_value = "2024-01-01T00:00:00Z"
-                        with patch("mcpgateway.bootstrap_db.logger") as mock_logger:
-                            await bootstrap_admin_user()
-
-                            mock_email_auth_service.create_user.assert_called_once_with(
-                                email=mock_settings.platform_admin_email, password=mock_settings.platform_admin_password, full_name=mock_settings.platform_admin_full_name, is_admin=True
-                            )
-                            assert mock_admin_user.email_verified_at == "2024-01-01T00:00:00Z"
-                            assert mock_db_session.commit.call_count == 2
-                            mock_logger.info.assert_any_call(f"Platform admin user created successfully: {mock_settings.platform_admin_email}")
+        with (
+            patch("mcpgateway.bootstrap_db.settings", mock_settings),
+            patch("mcpgateway.bootstrap_db.SessionLocal", return_value=mock_db_session),
+            patch("mcpgateway.services.email_auth_service.EmailAuthService", return_value=mock_email_auth_service),
+            patch("mcpgateway.db.utc_now") as mock_utc_now,
+            patch("mcpgateway.bootstrap_db.logger") as mock_logger,
+        ):
+            mock_utc_now.return_value = "2024-01-01T00:00:00Z"
+            await bootstrap_admin_user()
+
+            mock_email_auth_service.create_user.assert_called_once_with(
+                email=mock_settings.platform_admin_email,
+                password=mock_settings.platform_admin_password.get_secret_value(),
+                full_name=mock_settings.platform_admin_full_name,
+                is_admin=True,
+            )
+            assert mock_admin_user.email_verified_at == "2024-01-01T00:00:00Z"
+            assert mock_db_session.commit.call_count == 2
+            mock_logger.info.assert_any_call(f"Platform admin user created successfully: {mock_settings.platform_admin_email}")
 
     @pytest.mark.asyncio
     async def test_bootstrap_admin_user_with_personal_team(self, mock_settings, mock_db_session, mock_email_auth_service, mock_admin_user):

tests/unit/mcpgateway/test_config.py

@@ -11,12 +11,11 @@
 # Standard
 import os
 from pathlib import Path
-from typing import Any, Dict, List
 from unittest.mock import MagicMock, patch
 
-# Third-Party
-from fastapi import HTTPException
+from pydantic import SecretStr
 
+# Third-Party
 # Third-party
 import pytest
 
@@ -53,7 +52,7 @@ def test_parse_federation_peers_json_and_csv():
 # --------------------------------------------------------------------------- #
 #                          database / CORS helpers                            #
 # --------------------------------------------------------------------------- #
-def test_database_settings_sqlite_and_non_sqlite(tmp_path: Path):
+def test_database_settings_sqlite_and_non_sqlite(tmp_path: Path) -> None:
     """connect_args differs for sqlite vs everything else."""
     # sqlite -> check_same_thread flag present
     db_file = tmp_path / "foo" / "bar.db"
@@ -66,7 +65,7 @@ def test_database_settings_sqlite_and_non_sqlite(tmp_path: Path):
     assert s_pg.database_settings["connect_args"] == {}
 
 
-def test_validate_database_creates_missing_parent(tmp_path: Path):
+def test_validate_database_creates_missing_parent(tmp_path: Path) -> None:
     db_file = tmp_path / "newdir" / "db.sqlite"
     url = f"sqlite:///{db_file}"
     s = Settings(database_url=url, _env_file=None)
@@ -139,7 +138,7 @@ def test_settings_default_values():
         assert settings.port == 4444
         assert settings.database_url == "sqlite:///./mcp.db"
         assert settings.basic_auth_user == "admin"
-        assert settings.basic_auth_password == "changeme"
+        assert settings.basic_auth_password == SecretStr("changeme")
         assert settings.auth_required is True
         assert settings.jwt_secret_key.get_secret_value() == "x" * 32
         assert settings.auth_encryption_secret.get_secret_value() == "dummy-secret"

tests/unit/mcpgateway/test_coverage_push.py

@@ -13,6 +13,7 @@
 # Third-Party
 from fastapi import HTTPException
 from fastapi.testclient import TestClient
+from pydantic import SecretStr
 import pytest
 
 # First-Party
@@ -36,14 +37,14 @@ def test_require_api_key_scenarios():
     with patch("mcpgateway.main.settings") as mock_settings:
         mock_settings.auth_required = True
         mock_settings.basic_auth_user = "admin"
-        mock_settings.basic_auth_password = "secret"
+        mock_settings.basic_auth_password = SecretStr("secret")
         require_api_key("admin:secret")  # Should not raise
 
     # Test with auth enabled and incorrect key
     with patch("mcpgateway.main.settings") as mock_settings:
         mock_settings.auth_required = True
         mock_settings.basic_auth_user = "admin"
-        mock_settings.basic_auth_password = "secret"
+        mock_settings.basic_auth_password = SecretStr("secret")
 
         with pytest.raises(HTTPException):
             require_api_key("wrong:key")

tests/unit/mcpgateway/test_validate_env.py

@@ -1,20 +1,21 @@
 # -*- coding: utf-8 -*-
 # File: tests/unit/mcpgateway/test_validate_env.py
-from pathlib import Path
-import pytest
 import logging
 import os
+from pathlib import Path
 from unittest.mock import patch
 
-# Suppress mcpgateway.config logs during tests
-logging.getLogger("mcpgateway.config").setLevel(logging.ERROR)
+import pytest
 
 # Import the validate_env script directly
 from mcpgateway.scripts import validate_env as ve
 
+# Suppress mcpgateway.config logs during tests
+logging.getLogger("mcpgateway.config").setLevel(logging.ERROR)
+
 
 @pytest.fixture
-def valid_env(tmp_path: Path):
+def valid_env(tmp_path: Path) -> Path:
     envfile = tmp_path / ".env"
     envfile.write_text(
         "APP_DOMAIN=http://localhost:8000\n"
@@ -30,14 +31,14 @@ def valid_env(tmp_path: Path):
 
 
 @pytest.fixture
-def invalid_env(tmp_path: Path):
+def invalid_env(tmp_path: Path) -> Path:
     envfile = tmp_path / ".env"
     # Invalid URL + wrong log level + invalid port
     envfile.write_text("APP_DOMAIN=not-a-url\nPORT=-1\nLOG_LEVEL=wronglevel\n")
     return envfile
 
 
-def test_validate_env_success_direct(valid_env: Path):
+def test_validate_env_success_direct(valid_env: Path) -> None:
     """
     Test a valid .env. Warnings will be printed but do NOT fail the test.
     """
@@ -57,7 +58,7 @@ def test_validate_env_success_direct(valid_env: Path):
         assert code == 0
 
 
-def test_validate_env_failure_direct(invalid_env: Path):
+def test_validate_env_failure_direct(invalid_env: Path) -> None:
     """
     Test an invalid .env. Should fail due to ValidationError.
     """