Merge pull request #632 from Nayana-R-Gowda/425-SECURITYFEATURE

make test smoketest doctest passed

Author: madhav165

.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

@@ -60,7 +60,7 @@
 import jq
 from jsonpath_ng.ext import parse
 from jsonpath_ng.jsonpath import JSONPath
-from pydantic import field_validator
+from pydantic import Field, field_validator
 from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 
 logging.basicConfig(
@@ -116,6 +116,8 @@ class Settings(BaseSettings):
     auth_required: bool = True
     token_expiry: int = 10080  # minutes
 
+    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

@@ -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

@@ -17,6 +17,7 @@
     ...     basic_auth_user = 'user'
     ...     basic_auth_password = 'pass'
     ...     auth_required = True
+    ...     require_token_expiration = False
     >>> vc.settings = DummySettings()
     >>> import jwt
     >>> token = jwt.encode({'sub': 'alice'}, 'secret', algorithm='HS256')
@@ -39,6 +40,7 @@
 """
 
 # Standard
+import logging
 from typing import Optional
 
 # Third-Party
@@ -51,14 +53,16 @@
 )
 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)
 
+# Standard
+logger = logging.getLogger(__name__)
+
 
 async def verify_jwt_token(token: str) -> dict:
     """Verify and decode a JWT token.
@@ -74,6 +78,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
@@ -83,6 +88,7 @@ async def verify_jwt_token(token: str) -> dict:
         ...     basic_auth_user = 'user'
         ...     basic_auth_password = 'pass'
         ...     auth_required = True
+        ...     require_token_expiration = False
         >>> vc.settings = DummySettings()
         >>> import jwt
         >>> token = jwt.encode({'sub': 'alice'}, 'secret', algorithm='HS256')
@@ -108,22 +114,60 @@ 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
-        payload = jwt.decode(
-            token,
-            settings.jwt_secret_key,
-            algorithms=[settings.jwt_algorithm],
-            # options={"require": ["exp"]},  # Require expiration
+        # 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=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",
@@ -154,6 +198,7 @@ async def verify_credentials(token: str) -> dict:
         ...     basic_auth_user = 'user'
         ...     basic_auth_password = 'pass'
         ...     auth_required = True
+        ...     require_token_expiration = False
         >>> vc.settings = DummySettings()
         >>> import jwt
         >>> token = jwt.encode({'sub': 'alice'}, 'secret', algorithm='HS256')
@@ -194,6 +239,7 @@ async def require_auth(credentials: Optional[HTTPAuthorizationCredentials] = Dep
         ...     basic_auth_user = 'user'
         ...     basic_auth_password = 'pass'
         ...     auth_required = True
+        ...     require_token_expiration = False
         >>> vc.settings = DummySettings()
         >>> import jwt
         >>> from fastapi.security import HTTPAuthorizationCredentials
@@ -373,6 +419,7 @@ async def require_auth_override(
         ...     basic_auth_user = 'user'
         ...     basic_auth_password = 'pass'
         ...     auth_required = True
+        ...     require_token_expiration = False
         >>> vc.settings = DummySettings()
         >>> import jwt
         >>> import asyncio

tests/unit/mcpgateway/utils/test_verify_credentials.py

@@ -41,7 +41,15 @@
 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,6 +64,7 @@ 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)