fix: asymmetric jwt enocding and decoding

This commit introduces the option to load private and public keys to use also support asymmetric jwt encoding and decoding

Signed-off-by: Philip Miglinci <[email protected]>

Author: pmig

.dockerignore

@@ -241,6 +241,7 @@ log/
 
 # Certificates and secrets
 certs/
+jwt/
 *.pem
 *.key
 *.crt

.env.example

@@ -89,18 +89,26 @@ BASIC_AUTH_USER=admin
 BASIC_AUTH_PASSWORD=changeme
 AUTH_REQUIRED=true
 
+
+# Algorithm used to sign JWTs (e.g., HS256)
+JWT_ALGORITHM=HS256
+
 # Secret used to sign JWTs (use long random value in prod)
 # PRODUCTION: Use a strong, random secret (minimum 32 characters)
 JWT_SECRET_KEY=my-test-key
 
-# Algorithm used to sign JWTs (e.g., HS256)
-JWT_ALGORITHM=HS256
+# For asymetric algorithms liek RS256 public private keys are needed
+JWT_PUBLIC_KEY_PATH=jwt/public.pem
+JWT_PRIVATE_KEY_PATH=jwt/private.pem
 
 # JWT Audience and Issuer claims for token validation
 # PRODUCTION: Set these to your service-specific values
 JWT_AUDIENCE=mcpgateway-api
 JWT_ISSUER=mcpgateway
 
+# Disabling can be useful if dynamic client registration is used as the AUDIENCE in a JWT is the client
+JWT_AUDIENCE_VERIFICATION=true
+
 # Expiry time for generated JWT tokens (in minutes; e.g. 7 days)
 TOKEN_EXPIRY=10080
 REQUIRE_TOKEN_EXPIRATION=false

.gitignore

@@ -46,6 +46,7 @@ node_modules/
 .tmp*
 mcp.db-journal
 certs/
+jwt/
 FIXMEs
 *.old
 logs/

README.md

@@ -1040,36 +1040,39 @@ You can get started by copying the provided [.env.example](.env.example) to `.en
 
 ### Basic
 
-| Setting         | Description                              | Default                | Options                |
-| --------------- | ---------------------------------------- | ---------------------- | ---------------------- |
-| `APP_NAME`      | Gateway / OpenAPI title                  | `MCP Gateway`          | string                 |
-| `HOST`          | Bind address for the app                 | `127.0.0.1`            | IPv4/IPv6              |
-| `PORT`          | Port the server listens on               | `4444`                 | 1-65535                |
-| `DATABASE_URL`  | SQLAlchemy connection URL                | `sqlite:///./mcp.db`   | any SQLAlchemy dialect |
-| `APP_ROOT_PATH` | Subpath prefix for app (e.g. `/gateway`) | (empty)                | string                 |
-| `TEMPLATES_DIR` | Path to Jinja2 templates                 | `mcpgateway/templates` | path                   |
-| `STATIC_DIR`    | Path to static files                     | `mcpgateway/static`    | path                   |
-| `PROTOCOL_VERSION` | MCP protocol version supported          | `2025-03-26`           | string                 |
+| Setting            | Description                              | Default                | Options                |
+|--------------------|------------------------------------------|------------------------|------------------------|
+| `APP_NAME`         | Gateway / OpenAPI title                  | `MCP Gateway`          | string                 |
+| `HOST`             | Bind address for the app                 | `127.0.0.1`            | IPv4/IPv6              |
+| `PORT`             | Port the server listens on               | `4444`                 | 1-65535                |
+| `DATABASE_URL`     | SQLAlchemy connection URL                | `sqlite:///./mcp.db`   | any SQLAlchemy dialect |
+| `APP_ROOT_PATH`    | Subpath prefix for app (e.g. `/gateway`) | (empty)                | string                 |
+| `TEMPLATES_DIR`    | Path to Jinja2 templates                 | `mcpgateway/templates` | path                   |
+| `STATIC_DIR`       | Path to static files                     | `mcpgateway/static`    | path                   |
+| `PROTOCOL_VERSION` | MCP protocol version supported           | `2025-03-26`           | string                 |
 
 > 💡 Use `APP_ROOT_PATH=/foo` if reverse-proxying under a subpath like `https://host.com/foo/`.
 
 ### Authentication
 
-| Setting               | Description                                                      | Default       | Options    |
-| --------------------- | ---------------------------------------------------------------- | ------------- | ---------- |
-| `BASIC_AUTH_USER`     | Username for Admin UI login and HTTP Basic authentication        | `admin`       | string     |
-| `BASIC_AUTH_PASSWORD` | Password for Admin UI login and HTTP Basic authentication        | `changeme`    | string     |
-| `PLATFORM_ADMIN_EMAIL` | Email for bootstrap platform admin user (auto-created with admin privileges) | `[email protected]` | string |
-| `AUTH_REQUIRED`       | Require authentication for all API routes                        | `true`        | bool       |
-| `JWT_SECRET_KEY`      | Secret key used to **sign JWT tokens** for API access            | `my-test-key` | string     |
-| `JWT_ALGORITHM`       | Algorithm used to sign the JWTs (`HS256` is default, HMAC-based) | `HS256`       | PyJWT algs |
-| `JWT_AUDIENCE`        | JWT audience claim for token validation                           | `mcpgateway-api` | string  |
-| `JWT_ISSUER`          | JWT issuer claim for token validation                             | `mcpgateway`  | string     |
-| `TOKEN_EXPIRY`        | Expiry of generated JWTs in minutes                              | `10080`       | int > 0    |
-| `REQUIRE_TOKEN_EXPIRATION` | Require all JWT tokens to have expiration claims           | `false`       | bool       |
-| `AUTH_ENCRYPTION_SECRET` | Passphrase used to derive AES key for encrypting tool auth headers | `my-test-salt` | string |
-| `OAUTH_REQUEST_TIMEOUT` | OAuth request timeout in seconds                             | `30`          | int > 0    |
-| `OAUTH_MAX_RETRIES`   | Maximum retries for OAuth token requests                        | `3`           | int > 0    |
+| Setting                     | Description                                                                  | Default             | Options     |
+|-----------------------------|------------------------------------------------------------------------------|---------------------|-------------|
+| `BASIC_AUTH_USER`           | Username for Admin UI login and HTTP Basic authentication                    | `admin`             | string      |
+| `BASIC_AUTH_PASSWORD`       | Password for Admin UI login and HTTP Basic authentication                    | `changeme`          | string      |
+| `PLATFORM_ADMIN_EMAIL`      | Email for bootstrap platform admin user (auto-created with admin privileges) | `[email protected]` | string      |
+| `AUTH_REQUIRED`             | Require authentication for all API routes                                    | `true`              | bool        |
+| `JWT_ALGORITHM`             | Algorithm used to sign the JWTs (`HS256` is default, HMAC-based)             | `HS256`             | PyJWT algs  |
+| `JWT_SECRET_KEY`            | Secret key used to **sign JWT tok(empty)or API access                        | `my-test-key`       | string      |
+| `JWT_PUBLIC_KEY_PATH`       | If an asymmetric algorithm is used, a public key is required                 | (empty)             | path to pem |
+| `JWT_PRIVATE_KEY_PATH`      | If an asymmetric algorithm is used, a private key is required                | (empty)             | path to pem |
+| `JWT_AUDIENCE`              | JWT audience claim for token validation                                      | `mcpgateway-api`    | string      |
+| `JWT_AUDIENCE_VERIFICATION` | Disables jwt audience verification (useful for DCR)                          | `true`              | boolean     |
+| `JWT_ISSUER`                | JWT issuer claim for token validation                                        | `mcpgateway`        | string      |
+| `TOKEN_EXPIRY`              | Expiry of generated JWTs in minutes                                          | `10080`             | int > 0     |
+| `REQUIRE_TOKEN_EXPIRATION`  | Require all JWT tokens to have expiration claims                             | `false`             | bool        |
+| `AUTH_ENCRYPTION_SECRET`    | Passphrase used to derive AES key for encrypting tool auth headers           | `my-test-salt`      | string      |
+| `OAUTH_REQUEST_TIMEOUT`     | OAuth request timeout in seconds                                             | `30`                | int > 0     |
+| `OAUTH_MAX_RETRIES`         | Maximum retries for OAuth token requests                                     | `3`                 | int > 0     |
 
 > 🔐 `BASIC_AUTH_USER`/`PASSWORD` are used for:
 >

mcpgateway/admin.py

@@ -35,7 +35,6 @@
 from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, StreamingResponse
 import httpx
-import jwt
 from pydantic import ValidationError
 from pydantic_core import ValidationError as CoreValidationError
 from sqlalchemy.exc import IntegrityError
@@ -47,6 +46,7 @@
 from mcpgateway.db import Tool as DbTool
 from mcpgateway.middleware.rbac import get_current_user_with_permissions, require_permission
 from mcpgateway.models import LogLevel
+from mcpgateway.utils.create_jwt_token import create_jwt_token
 from mcpgateway.schemas import (
     A2AAgentCreate,
     GatewayCreate,
@@ -1890,7 +1890,8 @@ async def admin_ui(
                 "scopes": {"server_id": None, "permissions": ["*"], "ip_restrictions": [], "time_restrictions": {}},
             }
 
-            token = jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
+            # Generate token using centralized token creation
+            token = await create_jwt_token(payload)
 
             # Set HTTP-only cookie for security
             response.set_cookie(
@@ -2027,7 +2028,7 @@ async def admin_login_handler(request: Request, db: Session = Depends(get_db)) -
             # First-Party
             from mcpgateway.routers.email_auth import create_access_token  # pylint: disable=import-outside-toplevel
 
-            token, _ = create_access_token(user)  # expires_seconds not needed here
+            token, _ = await create_access_token(user)  # expires_seconds not needed here
 
             # Create redirect response
             root_path = request.scope.get("root_path", "")

mcpgateway/auth.py

@@ -14,12 +14,12 @@
 # Third-Party
 from fastapi import Depends, HTTPException, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
-import jwt
 from sqlalchemy.orm import Session
 
 # First-Party
 from mcpgateway.config import settings
 from mcpgateway.db import EmailUser, SessionLocal
+from mcpgateway.utils.verify_credentials import verify_jwt_token
 
 # Security scheme
 bearer_scheme = HTTPBearer(auto_error=False)
@@ -73,9 +73,9 @@ async def get_current_user(credentials: Optional[HTTPAuthorizationCredentials] =
     email = None
 
     try:
-        # Try JWT token first
+        # Try JWT token first using the centralized verify_jwt_token function
         logger.debug("Attempting JWT token validation")
-        payload = jwt.decode(credentials.credentials, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm], audience=settings.jwt_audience, issuer=settings.jwt_issuer)
+        payload = await verify_jwt_token(credentials.credentials)
 
         logger.debug("JWT token validated successfully")
         # Extract user identifier (support both new and legacy token formats)
@@ -113,13 +113,10 @@ async def get_current_user(credentials: Optional[HTTPAuthorizationCredentials] =
                 # Log the error but don't fail authentication for admin tokens
                 logger.warning(f"Token revocation check failed for JTI {jti}: {revoke_check_error}")
 
-    except jwt.ExpiredSignatureError:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Token expired",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-    except jwt.PyJWTError as jwt_error:
+    except HTTPException:
+        # Re-raise HTTPException from verify_jwt_token (handles expired/invalid tokens)
+        raise
+    except Exception as jwt_error:
         # JWT validation failed, try database API token
         logger.debug("JWT validation failed with error: %s, trying database API token", jwt_error)
         try:

mcpgateway/cache/session_registry.py

@@ -61,7 +61,6 @@
 
 # Third-Party
 from fastapi import HTTPException, status
-import jwt
 
 # First-Party
 from mcpgateway import __version__
@@ -71,6 +70,7 @@
 from mcpgateway.services import PromptService, ResourceService, ToolService
 from mcpgateway.services.logging_service import LoggingService
 from mcpgateway.transports import SSETransport
+from mcpgateway.utils.create_jwt_token import create_jwt_token
 from mcpgateway.utils.retry_manager import ResilientHttpClient
 from mcpgateway.validation.jsonrpc import JSONRPCError
 
@@ -1319,7 +1319,8 @@ async def generate_response(self, message: Dict[str, Any], transport: SSETranspo
                             "auth_provider": "internal",
                         },
                     }
-                    token = jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
+                    # Generate token using centralized token creation
+                    token = await create_jwt_token(payload)
 
                 headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
                 # Extract root URL from base_url (remove /servers/{id} path)

mcpgateway/config.py

@@ -163,10 +163,13 @@ class Settings(BaseSettings):
     # Authentication
     basic_auth_user: str = "admin"
     basic_auth_password: str = "changeme"
-    jwt_secret_key: str = "my-test-key"
     jwt_algorithm: str = "HS256"
-    jwt_audience: str = "mcpgateway-api"
-    jwt_issuer: str = "mcpgateway"
+    jwt_secret_key: str = "my-test-key"
+    jwt_public_key_path: str = ""
+    jwt_private_key_path: str = ""
+    jwt_audience: str = "JRKETZQLRBELYNGI5M3CUUNAL5"
+    jwt_issuer: str = "http://localhost:5556"
+    jwt_audience_verification: bool = True
     auth_required: bool = True
     token_expiry: int = 10080  # minutes
 

mcpgateway/middleware/token_scoping.py

@@ -19,11 +19,10 @@
 # Third-Party
 from fastapi import HTTPException, Request, status
 from fastapi.security import HTTPBearer
-import jwt
 
 # First-Party
-from mcpgateway.config import settings
 from mcpgateway.db import Permissions
+from mcpgateway.utils.verify_credentials import verify_jwt_token
 
 # Security scheme
 bearer_scheme = HTTPBearer(auto_error=False)
@@ -47,7 +46,7 @@ def __init__(self):
             True
         """
 
-    def _extract_token_scopes(self, request: Request) -> Optional[dict]:
+    async def _extract_token_scopes(self, request: Request) -> Optional[dict]:
         """Extract token scopes from JWT in request.
 
         Args:
@@ -64,11 +63,14 @@ def _extract_token_scopes(self, request: Request) -> Optional[dict]:
         token = auth_header.split(" ", 1)[1]
 
         try:
-            # Decode JWT with signature verification but skip audience/issuer checks for scope extraction
-            # (full verification including audience/issuer is handled by the auth system)
-            payload = jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm], options={"verify_aud": False, "verify_iss": False})
+            # Use the centralized verify_jwt_token function for consistent JWT validation
+            payload = await verify_jwt_token(token)
             return payload.get("scopes")
-        except jwt.PyJWTError:
+        except HTTPException:
+            # Token validation failed (expired, invalid, etc.)
+            return None
+        except Exception:
+            # Any other error in token validation
             return None
 
     def _get_client_ip(self, request: Request) -> str:
@@ -352,7 +354,7 @@ async def __call__(self, request: Request, call_next):
             return await call_next(request)
 
         # Extract token scopes
-        scopes = self._extract_token_scopes(request)
+        scopes = await self._extract_token_scopes(request)
 
         # If no scopes, continue (regular auth will handle this)
         if not scopes:

mcpgateway/routers/email_auth.py

@@ -24,14 +24,14 @@
 # Third-Party
 from fastapi import APIRouter, Depends, HTTPException, Request, status
 from fastapi.security import HTTPBearer
-import jwt
 from sqlalchemy.orm import Session
 
 # First-Party
 from mcpgateway.auth import get_current_user
 from mcpgateway.config import settings
 from mcpgateway.db import EmailUser, SessionLocal
 from mcpgateway.middleware.rbac import require_permission
+from mcpgateway.utils.create_jwt_token import create_jwt_token
 from mcpgateway.schemas import (
     AuthenticationResponse,
     AuthEventResponse,
@@ -104,7 +104,7 @@ def get_user_agent(request: Request) -> str:
     return request.headers.get("User-Agent", "unknown")
 
 
-def create_access_token(user: EmailUser, token_scopes: Optional[dict] = None, jti: Optional[str] = None) -> tuple[str, int]:
+async def create_access_token(user: EmailUser, token_scopes: Optional[dict] = None, jti: Optional[str] = None) -> tuple[str, int]:
     """Create JWT access token for user with enhanced scoping.
 
     Args:
@@ -149,13 +149,13 @@ def create_access_token(user: EmailUser, token_scopes: Optional[dict] = None, jt
         "scopes": token_scopes or {"server_id": None, "permissions": ["*"], "ip_restrictions": [], "time_restrictions": {}},  # Full access for regular user tokens
     }
 
-    # Generate token
-    token = jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
+    # Generate token using centralized token creation
+    token = await create_jwt_token(payload)
 
     return token, int(expires_delta.total_seconds())
 
 
-def create_legacy_access_token(user: EmailUser) -> tuple[str, int]:
+async def create_legacy_access_token(user: EmailUser) -> tuple[str, int]:
     """Create legacy JWT access token for backwards compatibility.
 
     Args:
@@ -181,8 +181,8 @@ def create_legacy_access_token(user: EmailUser) -> tuple[str, int]:
         "aud": settings.jwt_audience,
     }
 
-    # Generate token
-    token = jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
+    # Generate token using centralized token creation
+    token = await create_jwt_token(payload)
 
     return token, int(expires_delta.total_seconds())
 
@@ -226,7 +226,7 @@ async def login(login_request: EmailLoginRequest, request: Request, db: Session
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password")
 
         # Create access token
-        access_token, expires_in = create_access_token(user)
+        access_token, expires_in = await create_access_token(user)
 
         # Return authentication response
         return AuthenticationResponse(access_token=access_token, token_type="bearer", expires_in=expires_in, user=EmailUserResponse.from_email_user(user))
@@ -274,7 +274,7 @@ async def register(registration_request: EmailRegistrationRequest, request: Requ
         )
 
         # Create access token
-        access_token, expires_in = create_access_token(user)
+        access_token, expires_in = await create_access_token(user)
 
         logger.info(f"New user registered: {user.email}")