update

Signed-off-by: NAYANAR <[email protected]>

Author: NAYANAR0502

mcpgateway/config.py

@@ -60,8 +60,8 @@
 import jq
 from jsonpath_ng.ext import parse
 from jsonpath_ng.jsonpath import JSONPath
-from pydantic import Field,field_validator
-from pydantic_settings import BaseSettings,NoDecode,SettingsConfigDict 
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 
 logging.basicConfig(
     level=logging.INFO,
@@ -116,10 +116,7 @@ class Settings(BaseSettings):
     auth_required: bool = True
     token_expiry: int = 10080  # minutes
 
-    require_token_expiration: bool = Field(
-    default=False,  # Default to flexible mode for backward compatibility
-    description="Require all JWT tokens to have expiration claims"
-     )
+    require_token_expiration: bool = Field(default=False, description="Require all JWT tokens to have expiration claims")  # Default to flexible mode for backward compatibility
 
     #  Encryption key phrase for auth storage
     auth_encryption_secret: str = "my-test-salt"

mcpgateway/utils/create_jwt_token.py

@@ -111,8 +111,8 @@ def _create_jwt_token(
             "⚠️  WARNING: Creating token without expiration. This is a security risk!\n"
             "   Consider using --exp with a value > 0 for production use.\n"
             "   Once JWT API (#425) is available, use it for automatic token renewal.",
-            file=sys.stderr
-        )    
+            file=sys.stderr,
+        )
     return jwt.encode(payload, secret, algorithm=algorithm)
 
 

mcpgateway/utils/verify_credentials.py

@@ -39,6 +39,7 @@
 """
 
 # Standard
+import logging
 from typing import Optional
 
 # Third-Party
@@ -51,20 +52,18 @@
 )
 from fastapi.security.utils import get_authorization_scheme_param
 import jwt
-from jwt import PyJWTError
-
 
 # First-Party
 from mcpgateway.config import settings
 
 basic_security = HTTPBasic(auto_error=False)
 security = HTTPBearer(auto_error=False)
 
-import logging
+# Standard
 logger = logging.getLogger(__name__)
 
+
 async def verify_jwt_token(token: str) -> dict:
-    
     """Verify and decode a JWT token.
 
     Decodes and validates a JWT token using the configured secret key
@@ -78,6 +77,7 @@ async def verify_jwt_token(token: str) -> dict:
 
     Raises:
         HTTPException: 401 status if the token has expired or is invalid.
+        MissingRequiredClaimError: If the 'exp' claim is required but missing.
 
     Examples:
         >>> from mcpgateway.utils import verify_credentials as vc
@@ -143,23 +143,14 @@ async def verify_jwt_token(token: str) -> dict:
 
         # Log warning for non-expiring tokens
         if "exp" not in unverified:
-            logger.warning(
-                "JWT token without expiration accepted. "
-                "Consider enabling REQUIRE_TOKEN_EXPIRATION for better security. "
-                f"Token sub: {unverified.get('sub', 'unknown')}"
-            )
+            logger.warning("JWT token without expiration accepted. " "Consider enabling REQUIRE_TOKEN_EXPIRATION for better security. " f"Token sub: {unverified.get('sub', 'unknown')}")
 
         # Full validation
         options = {}
         if settings.require_token_expiration:
             options["require"] = ["exp"]
 
-        payload = jwt.decode(
-            token,
-            settings.jwt_secret_key,
-            algorithms=[settings.jwt_algorithm],
-            options=options
-        )
+        payload = jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm], options=options)
         return payload
 
     except jwt.MissingRequiredClaimError:
@@ -180,7 +171,8 @@ async def verify_jwt_token(token: str) -> dict:
             detail="Invalid token",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    
+
+
 async def verify_credentials(token: str) -> dict:
     """Verify credentials using a JWT token.
 

tests/security/test_rpc_endpoint_validation.py

@@ -39,14 +39,13 @@ class TestRPCEndpointValidation:
     @pytest.fixture
     def client(self):
         """Create a test client for the FastAPI app."""
-        return TestClient(app)   
+        return TestClient(app)
+
     @pytest.fixture
     def auth_headers(self):
         """Create authorization headers for testing."""
         # You might need to adjust this based on your auth setup
         return {"Authorization": "Bearer test-token", "Content-Type": "application/json"}
-    
-     
 
     def test_rpc_endpoint_with_malicious_methods(self, client, auth_headers):
         """Test that malicious method names are rejected before processing.

tests/unit/mcpgateway/utils/test_verify_credentials.py

@@ -48,6 +48,7 @@
 #         payload = payload | {"exp": int(expire.timestamp())}
 #     return jwt.encode(payload, secret, algorithm=ALGO)
 
+
 def _token(payload: dict, *, exp_delta: int | None = 60, secret: str = SECRET) -> str:
     """Return a signed JWT with optional expiry offset (minutes)."""
     if exp_delta is not None:
@@ -63,8 +64,8 @@ def _token(payload: dict, *, exp_delta: int | None = 60, secret: str = SECRET) -
 async def test_verify_jwt_token_success(monkeypatch):
     monkeypatch.setattr(vc.settings, "jwt_secret_key", SECRET, raising=False)
     monkeypatch.setattr(vc.settings, "jwt_algorithm", ALGO, raising=False)
-    monkeypatch.setattr(vc.settings, "require_token_expiration", False, raising=False) 
-    
+    monkeypatch.setattr(vc.settings, "require_token_expiration", False, raising=False)
+
     token = _token({"sub": "abc"})
     data = await vc.verify_jwt_token(token)