fixed tests

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

Author: NAYANAR0502

.env.example

@@ -97,6 +97,9 @@ JWT_ALGORITHM=HS256
 # Expiry time for generated JWT tokens (in minutes; e.g. 7 days)
 TOKEN_EXPIRY=10080
 
+# Require all JWT tokens to have expiration claims (true or false)
+REQUIRE_TOKEN_EXPIRATION=false
+
 # Used to derive an AES encryption key for secure auth storage
 # Must be a non-empty string (e.g. passphrase or random secret)
 AUTH_ENCRYPTION_SECRET=my-test-salt

mcpgateway/config.py

@@ -61,7 +61,8 @@
 from jsonpath_ng.ext import parse
 from jsonpath_ng.jsonpath import JSONPath
 from pydantic import field_validator
-from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
+from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict 
+from pydantic import Field
 
 logging.basicConfig(
     level=logging.INFO,
@@ -116,6 +117,11 @@ 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"
+     )
+
     #  Encryption key phrase for auth storage
     auth_encryption_secret: str = "my-test-salt"
 

mcpgateway/utils/create_jwt_token.py

@@ -105,6 +105,14 @@ def _create_jwt_token(
     if expires_in_minutes > 0:
         expire = _dt.datetime.now(_dt.timezone.utc) + _dt.timedelta(minutes=expires_in_minutes)
         payload["exp"] = int(expire.timestamp())
+    else:
+        # Warn about non-expiring token
+        print(
+            "⚠️  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
+        )    
     return jwt.encode(payload, secret, algorithm=algorithm)
 
 

mcpgateway/utils/verify_credentials.py

@@ -53,14 +53,18 @@
 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
+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
@@ -108,29 +112,75 @@ async def verify_jwt_token(token: str) -> dict:
         ...     print(e.status_code, e.detail)
         401 Invalid token
     """
+    # try:
+    #     Decode and validate token
+    #     payload = jwt.decode(
+    #         token,
+    #         settings.jwt_secret_key,
+    #         algorithms=[settings.jwt_algorithm],
+    #         # options={"require": ["exp"]},  # Require expiration
+    #     )
+    #     return payload  # Contains the claims (e.g., user info)
+    # except jwt.ExpiredSignatureError:
+    #     raise HTTPException(
+    #         status_code=status.HTTP_401_UNAUTHORIZED,
+    #         detail="Token has expired",
+    #         headers={"WWW-Authenticate": "Bearer"},
+    #     )
+    # except PyJWTError:
+    #     raise HTTPException(
+    #         status_code=status.HTTP_401_UNAUTHORIZED,
+    #         detail="Invalid token",
+    #         headers={"WWW-Authenticate": "Bearer"},
+    #     )
     try:
-        # Decode and validate token
+        # First decode to check claims
+        unverified = jwt.decode(token, options={"verify_signature": False})
+
+        # Check for expiration claim
+        if "exp" not in unverified and settings.require_token_expiration:
+            raise jwt.MissingRequiredClaimError("exp")
+
+        # 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')}"
+            )
+
+        # Full validation
+        options = {}
+        if settings.require_token_expiration:
+            options["require"] = ["exp"]
+
         payload = jwt.decode(
             token,
             settings.jwt_secret_key,
             algorithms=[settings.jwt_algorithm],
-            # options={"require": ["exp"]},  # Require expiration
+            options=options
+        )
+        return payload
+
+    except jwt.MissingRequiredClaimError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token is missing required expiration claim. Set REQUIRE_TOKEN_EXPIRATION=false to allow.",
+            headers={"WWW-Authenticate": "Bearer"},
         )
-        return payload  # Contains the claims (e.g., user info)
     except jwt.ExpiredSignatureError:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Token has expired",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    except PyJWTError:
+    except jwt.PyJWTError:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             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,13 +39,14 @@ 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

@@ -41,7 +41,14 @@
 ALGO = "HS256"
 
 
-def _token(payload: dict, *, exp_delta: int | None = None, secret: str = SECRET) -> str:
+# def _token(payload: dict, *, exp_delta: int | None = None, secret: str = SECRET) -> str:
+#     """Return a signed JWT with optional expiry offset (minutes)."""
+#     if exp_delta is not None:
+#         expire = datetime.now(timezone.utc) + timedelta(minutes=exp_delta)
+#         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:
         expire = datetime.now(timezone.utc) + timedelta(minutes=exp_delta)
@@ -56,7 +63,8 @@ def _token(payload: dict, *, exp_delta: int | None = None, 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) 
+    
     token = _token({"sub": "abc"})
     data = await vc.verify_jwt_token(token)