Refactor JSON functions from within config.py

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

Additional linting improvements

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

Lots of cleanup of secrets in code

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

fix: Correct Flake8 docstring errors in config.py

- Remove type annotations from parameter docstrings (DAR103)
- Add missing Raises sections for ValueError exceptions (DAR401)
- Fix return type descriptions to match TypedDict types (DAR203)

Resolves all DAR (docstring argument/return) validation errors.

Co-Authored-By: Claude <[email protected]>
Signed-off-by: Jonathan Springer <[email protected]>

Author: jonpspri

mcpgateway/bootstrap_db.py

@@ -78,7 +78,7 @@ async def bootstrap_admin_user() -> None:
             logger.info(f"Creating platform admin user: {settings.platform_admin_email}")
             admin_user = await auth_service.create_user(
                 email=settings.platform_admin_email,
-                password=settings.platform_admin_password,
+                password=settings.platform_admin_password.get_secret_value(),
                 full_name=settings.platform_admin_full_name,
                 is_admin=True,
             )

mcpgateway/config.py

@@ -45,6 +45,8 @@
 from fastapi.responses import JSONResponse, RedirectResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 from fastapi.templating import Jinja2Templates
+from jsonpath_ng.ext import parse
+from jsonpath_ng.jsonpath import JSONPath
 from pydantic import ValidationError
 from sqlalchemy import text
 from sqlalchemy.exc import IntegrityError
@@ -61,7 +63,7 @@
 from mcpgateway.auth import get_current_user
 from mcpgateway.bootstrap_db import main as bootstrap_db
 from mcpgateway.cache import ResourceCache, SessionRegistry
-from mcpgateway.config import jsonpath_modifier, settings
+from mcpgateway.config import settings
 from mcpgateway.db import refresh_slugs_on_startup, SessionLocal
 from mcpgateway.db import Tool as DbTool
 from mcpgateway.handlers.sampling import SamplingHandler
@@ -102,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
@@ -264,6 +265,75 @@ def get_user_email(user):
 resource_cache = ResourceCache(max_size=settings.resource_cache_size, ttl=settings.resource_cache_ttl)
 
 
+def jsonpath_modifier(data: Any, jsonpath: str = "$[*]", mappings: Optional[Dict[str, str]] = None) -> Union[List, Dict]:
+    """
+    Applies the given JSONPath expression and mappings to the data.
+    Only return data that is required by the user dynamically.
+
+    Args:
+        data: The JSON data to query.
+        jsonpath: The JSONPath expression to apply.
+        mappings: Optional dictionary of mappings where keys are new field names
+                  and values are JSONPath expressions.
+
+    Returns:
+        Union[List, Dict]: A list (or mapped list) or a Dict of extracted data.
+
+    Raises:
+        HTTPException: If there's an error parsing or executing the JSONPath expressions.
+
+    Examples:
+        >>> jsonpath_modifier({'a': 1, 'b': 2}, '$.a')
+        [1]
+        >>> jsonpath_modifier([{'a': 1}, {'a': 2}], '$[*].a')
+        [1, 2]
+        >>> jsonpath_modifier({'a': {'b': 2}}, '$.a.b')
+        [2]
+        >>> jsonpath_modifier({'a': 1}, '$.b')
+        []
+    """
+    if not jsonpath:
+        jsonpath = "$[*]"
+
+    try:
+        main_expr: JSONPath = parse(jsonpath)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=f"Invalid main JSONPath expression: {e}")
+
+    try:
+        main_matches = main_expr.find(data)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=f"Error executing main JSONPath: {e}")
+
+    results = [match.value for match in main_matches]
+
+    if mappings:
+        mapped_results = []
+        for item in results:
+            mapped_item = {}
+            for new_key, mapping_expr_str in mappings.items():
+                try:
+                    mapping_expr = parse(mapping_expr_str)
+                except Exception as e:
+                    raise HTTPException(status_code=400, detail=f"Invalid mapping JSONPath for key '{new_key}': {e}")
+                try:
+                    mapping_matches = mapping_expr.find(item)
+                except Exception as e:
+                    raise HTTPException(status_code=400, detail=f"Error executing mapping JSONPath for key '{new_key}': {e}")
+                if not mapping_matches:
+                    mapped_item[new_key] = None
+                elif len(mapping_matches) == 1:
+                    mapped_item[new_key] = mapping_matches[0].value
+                else:
+                    mapped_item[new_key] = [m.value for m in mapping_matches]
+            mapped_results.append(mapped_item)
+        results = mapped_results
+
+    if len(results) == 1 and isinstance(results[0], dict):
+        return results[0]
+    return results
+
+
 ####################
 # Startup/Shutdown #
 ####################
@@ -432,7 +502,7 @@ async def validate_security_configuration():
     if settings.jwt_secret_key == "my-test-key" and not settings.dev_mode:  # nosec B105 - checking for default value
         critical_issues.append("Using default JWT secret in non-dev mode. Set JWT_SECRET_KEY environment variable!")
 
-    if settings.basic_auth_password == "changeme" and settings.mcpgateway_ui_enabled:  # nosec B105 - checking for default value
+    if settings.basic_auth_password.get_secret_value() == "changeme" and settings.mcpgateway_ui_enabled:  # nosec B105 - checking for default value
         critical_issues.append("Admin UI enabled with default password. Set BASIC_AUTH_PASSWORD environment variable!")
 
     if not settings.auth_required and settings.federation_enabled and not settings.dev_mode:
@@ -469,7 +539,7 @@ async def validate_security_configuration():
             logger.info("  • Generate a strong JWT secret:")
             logger.info("    python3 -c 'import secrets; print(secrets.token_urlsafe(32))'")
 
-        if settings.basic_auth_password == "changeme":  # nosec B105 - checking for default value
+        if settings.basic_auth_password.get_secret_value() == "changeme":  # nosec B105 - checking for default value
             logger.info("  • Set a strong admin password in BASIC_AUTH_PASSWORD")
 
         if not settings.auth_required:
@@ -1011,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
@@ -1026,7 +1097,7 @@ def require_api_key(api_key: str) -> None:
         401
     """
     if settings.auth_required:
-        expected = f"{settings.basic_auth_user}:{settings.basic_auth_password}"
+        expected = f"{settings.basic_auth_user}:{settings.basic_auth_password.get_secret_value()}"
         if api_key != expected:
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key")
 
@@ -2623,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/main.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/scripts/validate_env.py

@@ -28,6 +28,7 @@
 
 # Third-Party
 import httpx
+import jq
 from mcp import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamablehttp_client
@@ -62,14 +63,61 @@
 from mcpgateway.utils.services_auth import decode_auth
 from mcpgateway.utils.sqlalchemy_modifier import json_contains_expr
 
-# Local
-from ..config import extract_using_jq
-
 # Initialize logging service first
 logging_service = LoggingService()
 logger = logging_service.get_logger(__name__)
 
 
+def extract_using_jq(data, jq_filter=""):
+    """
+    Extracts data from a given input (string, dict, or list) using a jq filter string.
+
+    Args:
+        data (str, dict, list): The input JSON data. Can be a string, dict, or list.
+        jq_filter (str): The jq filter string to extract the desired data.
+
+    Returns:
+        The result of applying the jq filter to the input data.
+
+    Examples:
+        >>> extract_using_jq('{"a": 1, "b": 2}', '.a')
+        [1]
+        >>> extract_using_jq({'a': 1, 'b': 2}, '.b')
+        [2]
+        >>> extract_using_jq('[{"a": 1}, {"a": 2}]', '.[].a')
+        [1, 2]
+        >>> extract_using_jq('not a json', '.a')
+        ['Invalid JSON string provided.']
+        >>> extract_using_jq({'a': 1}, '')
+        {'a': 1}
+    """
+    if jq_filter == "":
+        return data
+    if isinstance(data, str):
+        # If the input is a string, parse it as JSON
+        try:
+            data = json.loads(data)
+        except json.JSONDecodeError:
+            return ["Invalid JSON string provided."]
+
+    elif not isinstance(data, (dict, list)):
+        # If the input is not a string, dict, or list, raise an error
+        return ["Input data must be a JSON string, dictionary, or list."]
+
+    # Apply the jq filter to the data
+    try:
+        # Pylint can't introspect C-extension modules, so it doesn't know that jq really does export an all() function.
+        # pylint: disable=c-extension-no-member
+        result = jq.all(jq_filter, data)  # Use `jq.all` to get all matches (returns a list)
+        if result == [None]:
+            result = "Error applying jsonpath filter"
+    except Exception as e:
+        message = "Error applying jsonpath filter: " + str(e)
+        return message
+
+    return result
+
+
 class ToolError(Exception):
     """Base class for tool-related errors.
 

mcpgateway/services/tool_service.py

@@ -117,7 +117,7 @@ def get_predefined_sso_providers() -> List[Dict]:
                 "display_name": "GitHub",
                 "provider_type": "oauth2",
                 "client_id": settings.sso_github_client_id,
-                "client_secret": settings.sso_github_client_secret or "",
+                "client_secret": settings.sso_github_client_secret.get_secret_value() if settings.sso_github_client_secret else "",
                 "authorization_url": "https://github.com/login/oauth/authorize",
                 "token_url": "https://github.com/login/oauth/access_token",
                 "userinfo_url": "https://api.github.com/user",
@@ -137,7 +137,7 @@ def get_predefined_sso_providers() -> List[Dict]:
                 "display_name": "Google",
                 "provider_type": "oidc",
                 "client_id": settings.sso_google_client_id,
-                "client_secret": settings.sso_google_client_secret or "",
+                "client_secret": settings.sso_google_client_secret.get_secret_value() if settings.sso_google_client_secret else "",
                 "authorization_url": "https://accounts.google.com/o/oauth2/auth",
                 "token_url": "https://oauth2.googleapis.com/token",
                 "userinfo_url": "https://openidconnect.googleapis.com/v1/userinfo",
@@ -159,7 +159,7 @@ def get_predefined_sso_providers() -> List[Dict]:
                 "display_name": "IBM Security Verify",
                 "provider_type": "oidc",
                 "client_id": settings.sso_ibm_verify_client_id,
-                "client_secret": settings.sso_ibm_verify_client_secret or "",
+                "client_secret": settings.sso_ibm_verify_client_secret.get_secret_value() if settings.sso_ibm_verify_client_secret else "",
                 "authorization_url": f"{base_url}/oidc/endpoint/default/authorize",
                 "token_url": f"{base_url}/oidc/endpoint/default/token",
                 "userinfo_url": f"{base_url}/oidc/endpoint/default/userinfo",
@@ -181,7 +181,7 @@ def get_predefined_sso_providers() -> List[Dict]:
                 "display_name": "Okta",
                 "provider_type": "oidc",
                 "client_id": settings.sso_okta_client_id,
-                "client_secret": settings.sso_okta_client_secret or "",
+                "client_secret": settings.sso_okta_client_secret.get_secret_value() if settings.sso_okta_client_secret else "",
                 "authorization_url": f"{base_url}/oauth2/default/v1/authorize",
                 "token_url": f"{base_url}/oauth2/default/v1/token",
                 "userinfo_url": f"{base_url}/oauth2/default/v1/userinfo",
@@ -204,7 +204,7 @@ def get_predefined_sso_providers() -> List[Dict]:
                 "display_name": "Microsoft Entra ID",
                 "provider_type": "oidc",
                 "client_id": settings.sso_entra_client_id,
-                "client_secret": settings.sso_entra_client_secret or "",
+                "client_secret": settings.sso_entra_client_secret.get_secret_value() if settings.sso_entra_client_secret else "",
                 "authorization_url": f"{base_url}/oauth2/v2.0/authorize",
                 "token_url": f"{base_url}/oauth2/v2.0/token",
                 "userinfo_url": "https://graph.microsoft.com/oidc/userinfo",
@@ -232,7 +232,7 @@ def get_predefined_sso_providers() -> List[Dict]:
                         "display_name": f"Keycloak ({settings.sso_keycloak_realm})",
                         "provider_type": "oidc",
                         "client_id": settings.sso_keycloak_client_id,
-                        "client_secret": settings.sso_keycloak_client_secret or "",
+                        "client_secret": settings.sso_keycloak_client_secret.get_secret_value() if settings.sso_keycloak_client_secret else "",
                         "authorization_url": endpoints["authorization_url"],
                         "token_url": endpoints["token_url"],
                         "userinfo_url": endpoints["userinfo_url"],
@@ -270,7 +270,7 @@ def get_predefined_sso_providers() -> List[Dict]:
                 "display_name": display_name,
                 "provider_type": "oidc",
                 "client_id": settings.sso_generic_client_id,
-                "client_secret": settings.sso_generic_client_secret or "",
+                "client_secret": settings.sso_generic_client_secret.get_secret_value() if settings.sso_generic_client_secret else "",
                 "authorization_url": settings.sso_generic_authorization_url,
                 "token_url": settings.sso_generic_token_url,
                 "userinfo_url": settings.sso_generic_userinfo_url,