JWT (JSON Web Token): Personal Journey with MS Entra Integration

Introduction

In my journey as a developer working with modern authentication systems, I've encountered numerous scenarios where understanding JWT (JSON Web Token) has been crucial. This post shares my personal experience and knowledge about JWT tokens, their integration with Microsoft Entra (formerly Azure AD), and practical Python implementations.

What is JWT (JSON Web Token)?

JWT is a compact, URL-safe means of representing claims to be transferred between two parties. Think of it as a digital passport that contains information about a user and can be verified without needing to contact the issuing authority every time.

JWT Structure

A JWT consists of three parts separated by dots (.):

eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiYWRtaW4iOnRydWV9.EkN-DOsnsuRjRO6BxXemmJDm3HbxrbRzXglbN2S4sOkopdU4IsDxTI8jO19W_A4K8ZPJijNLis4EZsHeY559a4DFOd50_OqgHs3PH-S7wRL0isHoN7vX-XvJ1VGH_8aS7XYQCmHQJL_OYTrOL_ZQ

1. Header

Contains metadata about the token:

{
  "alg": "RS256",
  "typ": "JWT",
  "kid": "key-id"
}

2. Payload (Claims)

Contains the actual data/claims:

{
  "sub": "user-id",
  "name": "John Doe",
  "email": "[email protected]",
  "roles": ["user", "admin"],
  "iat": 1516239022,
  "exp": 1516242622
}

3. Signature

Ensures the token hasn't been tampered with:

RSASHA256(
  base64UrlEncode(header) + "." +
  base64UrlEncode(payload),
  secret)

How JWT Works

From my experience implementing JWT-based authentication systems, here's how the flow typically works:

1. User Authentication: User provides credentials to the Identity Provider (MS Entra)

2. Token Issuance: MS Entra validates credentials and issues a JWT

3. Token Usage: Client includes JWT in API requests

4. Token Verification: API server validates the JWT signature and claims

5. Access Decision: Based on token validity and claims, access is granted or denied

JWT Authentication Flow Diagram

The following sequence diagram illustrates the complete JWT authentication flow with MS Entra:

Authentication Flow Details

Phase 1: Initial Authentication

• Steps 1-3: User provides credentials to the client application, which forwards them to MS Entra

• Steps 4-5: MS Entra validates credentials and issues a JWT containing user claims

Phase 2: API Access with JWT

• Step 6: Client includes JWT in the Authorization header (Bearer <token>)

• Step 7: API server extracts the JWT from the request header

Phase 3: Token Validation

• Steps 8-9: API server retrieves public keys from MS Entra's JWKS endpoint (cached for performance)

• Steps 10-11: Server validates the JWT signature and verifies standard claims

• Steps 12-14: Based on validation results, server either grants access or returns appropriate error codes

Phase 4: Token Refresh (When Needed)

• Steps 15-18: When tokens expire, the refresh flow allows obtaining new tokens without re-authentication

Pros and Cons of JWT

Pros (From My Experience)

1. Stateless Authentication

• No need to store session data on the server

• Scales horizontally without session affinity

• Perfect for microservices architecture

2. Cross-Domain Support

• Works seamlessly across different domains

• Ideal for Single Sign-On (SSO) implementations

• Great for mobile and web applications

3. Rich Information Storage

• Can contain user roles, permissions, and custom claims

• Reduces database lookups for user information

• Self-contained authorization data

4. Industry Standard

• Widely supported across platforms and languages

• RFC 7519 specification ensures consistency

• Large ecosystem of libraries and tools

Cons (Lessons Learned)

1. Token Size

• Larger than simple session IDs

• Can impact network performance with many claims

• HTTP header size limitations

2. Token Revocation Challenges

• Cannot easily revoke tokens before expiration

• Requires additional mechanisms for immediate revocation

• Blacklisting adds complexity

3. Security Considerations

• Sensitive data exposure if not properly secured

• XSS vulnerabilities if stored in localStorage

• Requires careful key management

4. Token Refresh Complexity

• Need to handle token expiration gracefully

• Refresh token implementation adds complexity

• Clock skew issues between systems

MS Entra as Identity Provider

Microsoft Entra (formerly Azure AD) serves as an excellent Identity Provider for JWT-based authentication. Here's why I prefer it:

Key Benefits:

• Enterprise Integration: Seamless integration with Microsoft ecosystem

• Security Features: Multi-factor authentication, conditional access

• Compliance: SOC 2, ISO 27001, and other certifications

• Scalability: Handles millions of users and applications

MS Entra JWT Token Claims

Common claims in MS Entra tokens:

• aud: Audience (your application ID)

• iss: Issuer (MS Entra tenant)

• sub: Subject (user identifier)

• upn: User Principal Name

• roles: Application roles

• groups: Security groups

Python Implementation: Getting JWT from MS Entra

Here's a practical Python implementation I've used in production:

Prerequisites

First, install required packages:

pip install msal requests PyJWT cryptography

1. Application Registration Setup

# config.py
import os

# MS Entra App Registration Details
CLIENT_ID = os.getenv('AZURE_CLIENT_ID', 'your-client-id')
CLIENT_SECRET = os.getenv('AZURE_CLIENT_SECRET', 'your-client-secret')
TENANT_ID = os.getenv('AZURE_TENANT_ID', 'your-tenant-id')
SCOPE = ['https://graph.microsoft.com/.default']

# MS Entra Endpoints
AUTHORITY = f'https://login.microsoftonline.com/{TENANT_ID}'

2. JWT Token Acquisition

# jwt_client.py
import msal
import requests
import jwt
from typing import Dict, Optional
from config import CLIENT_ID, CLIENT_SECRET, TENANT_ID, AUTHORITY, SCOPE

class MSEntraJWTClient:
    def __init__(self):
        self.app = msal.ConfidentialClientApplication(
            CLIENT_ID,
            authority=AUTHORITY,
            client_credential=CLIENT_SECRET,
        )
        self.token_cache = {}
    
    def get_access_token(self) -> Optional[str]:
        """
        Get access token from MS Entra using client credentials flow
        """
        try:
            # Try to get token from cache first
            result = self.app.acquire_token_silent(SCOPE, account=None)
            
            if not result:
                # Get new token
                result = self.app.acquire_token_for_client(scopes=SCOPE)
            
            if "access_token" in result:
                return result["access_token"]
            else:
                print(f"Error getting token: {result.get('error_description')}")
                return None
                
        except Exception as e:
            print(f"Exception getting token: {str(e)}")
            return None
    
    def get_user_token(self, username: str, password: str) -> Optional[str]:
        """
        Get user token using Resource Owner Password Credentials (ROPC) flow
        Note: ROPC is not recommended for production use
        """
        try:
            result = self.app.acquire_token_by_username_password(
                username=username,
                password=password,
                scopes=SCOPE
            )
            
            if "access_token" in result:
                return result["access_token"]
            else:
                print(f"Error getting user token: {result.get('error_description')}")
                return None
                
        except Exception as e:
            print(f"Exception getting user token: {str(e)}")
            return None

3. JWT Token Validation and Claims Extraction

# jwt_validator.py
import jwt
import requests
from typing import Dict, Optional
from cryptography.hazmat.primitives import serialization
from config import TENANT_ID, CLIENT_ID

class JWTValidator:
    def __init__(self):
        self.jwks_url = f'https://login.microsoftonline.com/{TENANT_ID}/discovery/v2.0/keys'
        self.issuer = f'https://login.microsoftonline.com/{TENANT_ID}/v2.0'
        self.audience = CLIENT_ID
        self._jwks_cache = None
    
    def _get_jwks(self) -> Dict:
        """Get JSON Web Key Set from MS Entra"""
        if not self._jwks_cache:
            response = requests.get(self.jwks_url)
            response.raise_for_status()
            self._jwks_cache = response.json()
        return self._jwks_cache
    
    def _get_signing_key(self, token_header: Dict) -> str:
        """Get the signing key for token validation"""
        jwks = self._get_jwks()
        
        for key in jwks['keys']:
            if key['kid'] == token_header['kid']:
                # Convert JWK to PEM format
                key_obj = jwt.algorithms.RSAAlgorithm.from_jwk(key)
                return key_obj.public_key().public_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PublicFormat.SubjectPublicKeyInfo
                )
        
        raise jwt.InvalidKeyError(f"Unable to find key with kid: {token_header['kid']}")
    
    def validate_token(self, token: str) -> Optional[Dict]:
        """
        Validate JWT token and extract claims
        """
        try:
            # Decode token header to get key ID
            unverified_header = jwt.get_unverified_header(token)
            
            # Get signing key
            signing_key = self._get_signing_key(unverified_header)
            
            # Validate and decode token
            decoded_token = jwt.decode(
                token,
                signing_key,
                algorithms=['RS256'],
                audience=self.audience,
                issuer=self.issuer,
                options={"verify_exp": True}
            )
            
            return decoded_token
            
        except jwt.ExpiredSignatureError:
            print("Token has expired")
            return None
        except jwt.InvalidAudienceError:
            print("Invalid audience")
            return None
        except jwt.InvalidIssuerError:
            print("Invalid issuer")
            return None
        except jwt.InvalidSignatureError:
            print("Invalid signature")
            return None
        except Exception as e:
            print(f"Token validation error: {str(e)}")
            return None
    
    def extract_user_claims(self, token: str) -> Optional[Dict]:
        """
        Extract user-specific claims from JWT token
        """
        claims = self.validate_token(token)
        if not claims:
            return None
        
        user_claims = {
            'user_id': claims.get('sub'),
            'username': claims.get('preferred_username'),
            'name': claims.get('name'),
            'email': claims.get('email'),
            'roles': claims.get('roles', []),
            'groups': claims.get('groups', []),
            'tenant_id': claims.get('tid'),
            'app_id': claims.get('aud'),
            'issued_at': claims.get('iat'),
            'expires_at': claims.get('exp')
        }
        
        return user_claims

4. Application Integration Example

# app.py
from flask import Flask, request, jsonify
from jwt_client import MSEntraJWTClient
from jwt_validator import JWTValidator
from functools import wraps

app = Flask(__name__)
jwt_client = MSEntraJWTClient()
jwt_validator = JWTValidator()

def require_auth(f):
    """Decorator to require valid JWT token"""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        auth_header = request.headers.get('Authorization')
        
        if not auth_header or not auth_header.startswith('Bearer '):
            return jsonify({'error': 'Missing or invalid Authorization header'}), 401
        
        token = auth_header.split(' ')[1]
        claims = jwt_validator.validate_token(token)
        
        if not claims:
            return jsonify({'error': 'Invalid token'}), 401
        
        # Add claims to request context
        request.user_claims = claims
        return f(*args, **kwargs)
    
    return decorated_function

def require_role(required_role):
    """Decorator to require specific role"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            user_roles = request.user_claims.get('roles', [])
            
            if required_role not in user_roles:
                return jsonify({'error': 'Insufficient permissions'}), 403
            
            return f(*args, **kwargs)
        return decorated_function
    return decorator

@app.route('/login', methods=['POST'])
def login():
    """Login endpoint to get JWT token"""
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    
    if not username or not password:
        return jsonify({'error': 'Username and password required'}), 400
    
    token = jwt_client.get_user_token(username, password)
    
    if token:
        user_claims = jwt_validator.extract_user_claims(token)
        return jsonify({
            'access_token': token,
            'user_claims': user_claims
        })
    else:
        return jsonify({'error': 'Authentication failed'}), 401

@app.route('/profile', methods=['GET'])
@require_auth
def get_profile():
    """Get user profile from JWT claims"""
    user_claims = jwt_validator.extract_user_claims(
        request.headers.get('Authorization').split(' ')[1]
    )
    
    return jsonify({
        'profile': user_claims
    })

@app.route('/admin', methods=['GET'])
@require_auth
@require_role('Admin')
def admin_endpoint():
    """Admin-only endpoint"""
    return jsonify({
        'message': 'Welcome to admin area',
        'user': request.user_claims.get('preferred_username')
    })

if __name__ == '__main__':
    app.run(debug=True)

5. Usage Example

# example_usage.py
from jwt_client import MSEntraJWTClient
from jwt_validator import JWTValidator

def main():
    # Initialize clients
    jwt_client = MSEntraJWTClient()
    jwt_validator = JWTValidator()
    
    # Get application token (client credentials)
    app_token = jwt_client.get_access_token()
    if app_token:
        print("Application Token acquired successfully!")
        
        # Validate and extract claims
        claims = jwt_validator.validate_token(app_token)
        if claims:
            print(f"Token valid. App ID: {claims.get('aud')}")
            print(f"Expires at: {claims.get('exp')}")
        else:
            print("Token validation failed")
    
    # Example: Get user token (for demo - use proper auth flow in production)
    user_token = jwt_client.get_user_token('[email protected]', 'password')
    if user_token:
        print("User Token acquired successfully!")
        
        # Extract user claims
        user_claims = jwt_validator.extract_user_claims(user_token)
        if user_claims:
            print(f"User: {user_claims.get('name')}")
            print(f"Email: {user_claims.get('email')}")
            print(f"Roles: {user_claims.get('roles')}")

if __name__ == '__main__':
    main()

JWT Debugging and Validation Tools

During my development journey with JWT tokens, I've found several online tools invaluable for debugging, testing, and understanding JWT tokens. Here are the most useful ones:

1. JWT.io - The Primary JWT Debugger

JWT.io is the go-to tool for JWT debugging and is maintained by Auth0. Here's how I use it in my development workflow:

Key Features:

• Token Decoding: Paste any JWT to see its decoded header, payload, and signature

• Real-time Validation: Instantly see if a token is valid or expired

• Algorithm Support: Supports all standard JWT algorithms (HS256, RS256, etc.)

• Key Verification: Test signature validation with public/private keys

Practical Use Cases from My Experience:

A. Debugging MS Entra Tokens

// Example: Debugging an MS Entra token
// 1. Copy your JWT token from the browser/application
// 2. Paste it into jwt.io
// 3. Examine the claims structure

// Typical MS Entra token payload you'll see:
{
  "aud": "your-client-id",
  "iss": "https://login.microsoftonline.com/tenant-id/v2.0",
  "iat": 1641234567,
  "nbf": 1641234567,
  "exp": 1641238167,
  "sub": "user-object-id",
  "tid": "tenant-id",
  "upn": "[email protected]",
  "name": "John Doe",
  "preferred_username": "[email protected]",
  "roles": ["User", "Admin"],
  "groups": ["group-id-1", "group-id-2"]
}

B. Signature Verification

# To verify MS Entra token signature on jwt.io:
# 1. Get the public key from MS Entra JWKS endpoint
curl https://login.microsoftonline.com/{tenant-id}/discovery/v2.0/keys

# 2. Copy the appropriate public key (matching 'kid' from token header)
# 3. Paste it in the "Verify Signature" section on jwt.io
# 4. The signature will show as verified if valid

2. Other Useful JWT Tools

A. Token.dev

• URL: token.dev

• Speciality: Advanced JWT analysis with security insights

• Best For: Security auditing and compliance checking

B. JWT Tool (Chrome Extension)

• Feature: Browser-based JWT debugging

• Best For: Debugging tokens directly in browser developer tools

C. Postman JWT Debugger

• Feature: Built-in JWT debugging in Postman

• Best For: API testing workflows with JWT authentication

3. Python Script for JWT Analysis

Here's a Python script I use for automated JWT analysis and validation:

# jwt_analyzer.py
import jwt
import json
import requests
from datetime import datetime
from typing import Dict, Any

class JWTAnalyzer:
    def __init__(self):
        self.analysis_results = {}
    
    def analyze_token(self, token: str) -> Dict[str, Any]:
        """
        Comprehensive JWT token analysis
        """
        analysis = {
            'token': token,
            'header': {},
            'payload': {},
            'signature_info': {},
            'validation_results': {},
            'security_analysis': {}
        }
        
        try:
            # Decode header without verification
            analysis['header'] = jwt.get_unverified_header(token)
            
            # Decode payload without verification
            analysis['payload'] = jwt.decode(token, options={"verify_signature": False})
            
            # Analyze token structure
            analysis['signature_info'] = self._analyze_signature(analysis['header'])
            
            # Validate token timing
            analysis['validation_results'] = self._validate_timing(analysis['payload'])
            
            # Security analysis
            analysis['security_analysis'] = self._security_analysis(analysis['payload'])
            
            return analysis
            
        except Exception as e:
            analysis['error'] = str(e)
            return analysis
    
    def _analyze_signature(self, header: Dict) -> Dict:
        """Analyze JWT signature information"""
        return {
            'algorithm': header.get('alg', 'Unknown'),
            'token_type': header.get('typ', 'Unknown'),
            'key_id': header.get('kid', 'Not specified'),
            'is_symmetric': header.get('alg', '').startswith('HS'),
            'is_asymmetric': header.get('alg', '').startswith('RS') or header.get('alg', '').startswith('ES')
        }
    
    def _validate_timing(self, payload: Dict) -> Dict:
        """Validate token timing claims"""
        now = datetime.utcnow().timestamp()
        
        results = {
            'current_time': now,
            'is_expired': False,
            'is_not_before_valid': True,
            'time_to_expiry': None
        }
        
        # Check expiration
        if 'exp' in payload:
            exp_time = payload['exp']
            results['expiry_time'] = exp_time
            results['is_expired'] = now > exp_time
            results['time_to_expiry'] = exp_time - now if not results['is_expired'] else 0
        
        # Check not before
        if 'nbf' in payload:
            nbf_time = payload['nbf']
            results['not_before_time'] = nbf_time
            results['is_not_before_valid'] = now >= nbf_time
        
        # Check issued at
        if 'iat' in payload:
            results['issued_at'] = payload['iat']
            results['token_age'] = now - payload['iat']
        
        return results
    
    def _security_analysis(self, payload: Dict) -> Dict:
        """Perform security analysis on JWT claims"""
        warnings = []
        recommendations = []
        
        # Check for sensitive data in payload
        sensitive_fields = ['password', 'secret', 'key', 'token']
        for field in payload.keys():
            if any(sensitive in field.lower() for sensitive in sensitive_fields):
                warnings.append(f"Potentially sensitive field '{field}' found in payload")
        
        # Check token expiry duration
        if 'exp' in payload and 'iat' in payload:
            duration = payload['exp'] - payload['iat']
            if duration > 86400:  # More than 24 hours
                warnings.append(f"Long token expiry duration: {duration/3600:.1f} hours")
            elif duration > 3600:  # More than 1 hour
                recommendations.append(f"Consider shorter token expiry: {duration/3600:.1f} hours")
        
        # Check for required claims
        required_claims = ['aud', 'iss', 'exp', 'sub']
        missing_claims = [claim for claim in required_claims if claim not in payload]
        if missing_claims:
            warnings.append(f"Missing recommended claims: {missing_claims}")
        
        return {
            'warnings': warnings,
            'recommendations': recommendations,
            'claim_count': len(payload),
            'has_roles': 'roles' in payload,
            'has_groups': 'groups' in payload
        }
    
    def print_analysis(self, token: str):
        """Print formatted analysis results"""
        analysis = self.analyze_token(token)
        
        print("=" * 80)
        print("JWT TOKEN ANALYSIS")
        print("=" * 80)
        
        # Header Analysis
        print("\n📋 HEADER ANALYSIS:")
        header = analysis['header']
        print(f"  Algorithm: {header.get('alg', 'Unknown')}")
        print(f"  Type: {header.get('typ', 'Unknown')}")
        print(f"  Key ID: {header.get('kid', 'Not specified')}")
        
        # Payload Analysis
        print("\n📦 PAYLOAD ANALYSIS:")
        payload = analysis['payload']
        print(f"  Issuer: {payload.get('iss', 'Not specified')}")
        print(f"  Audience: {payload.get('aud', 'Not specified')}")
        print(f"  Subject: {payload.get('sub', 'Not specified')}")
        print(f"  User: {payload.get('name', payload.get('preferred_username', 'Not specified'))}")
        
        # Timing Analysis
        print("\n⏰ TIMING ANALYSIS:")
        timing = analysis['validation_results']
        if timing.get('is_expired'):
            print("  ❌ Token is EXPIRED")
        else:
            if timing.get('time_to_expiry'):
                print(f"  ✅ Token expires in {timing['time_to_expiry']/3600:.1f} hours")
        
        # Security Analysis
        print("\n🔒 SECURITY ANALYSIS:")
        security = analysis['security_analysis']
        if security['warnings']:
            print("  ⚠️  WARNINGS:")
            for warning in security['warnings']:
                print(f"    - {warning}")
        
        if security['recommendations']:
            print("  💡 RECOMMENDATIONS:")
            for rec in security['recommendations']:
                print(f"    - {rec}")
        
        print(f"\n📊 SUMMARY:")
        print(f"  Total Claims: {security['claim_count']}")
        print(f"  Has Roles: {'✅' if security['has_roles'] else '❌'}")
        print(f"  Has Groups: {'✅' if security['has_groups'] else '❌'}")
        print("=" * 80)

# Usage Example
if __name__ == '__main__':
    analyzer = JWTAnalyzer()
    
    # Example token (replace with your actual token)
    sample_token = "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsIng1dCI6Im..."
    
    # Analyze the token
    analyzer.print_analysis(sample_token)

4. Best Practices for JWT Debugging

Based on my experience, here are essential practices when working with JWT tokens:

A. Development Workflow

1. Use jwt.io for immediate debugging when tokens don't work as expected

2. Verify token structure matches your application's expectations

3. Check expiration times to ensure tokens aren't expired during testing

4. Validate audience and issuer claims match your application configuration

B. Production Debugging

1. Never paste production tokens into online tools (security risk)

2. Use local debugging tools or create sanitized test tokens

3. Implement proper logging for token validation failures

4. Monitor token usage patterns for anomalies

C. Security Considerations

# Example: Safe token debugging function
def debug_token_safely(token: str, mask_sensitive: bool = True) -> Dict:
    """
    Debug JWT token with option to mask sensitive information
    """
    try:
        payload = jwt.decode(token, options={"verify_signature": False})
        
        if mask_sensitive:
            # Mask sensitive fields for logging
            sensitive_fields = ['sub', 'email', 'upn', 'oid']
            for field in sensitive_fields:
                if field in payload:
                    payload[field] = f"***{payload[field][-4:]}" if len(payload[field]) > 4 else "***"
        
        return {
            'status': 'success',
            'payload': payload,
            'expires_at': datetime.fromtimestamp(payload.get('exp', 0)).isoformat(),
            'issued_at': datetime.fromtimestamp(payload.get('iat', 0)).isoformat()
        }
    except Exception as e:
        return {
            'status': 'error',
            'error': str(e)
        }

5. Common JWT Issues and Debugging Tips

From my troubleshooting experience, here are common issues and how to debug them:

A. Token Validation Failures

# Issue: "Invalid signature" error
# Debug steps:
# 1. Check if you're using the correct public key
# 2. Verify the key ID (kid) matches between token and JWKS
# 3. Ensure algorithm matches (RS256 vs HS256)

# Use this command to check MS Entra JWKS:
curl -s "https://login.microsoftonline.com/{tenant-id}/discovery/v2.0/keys" | jq '.keys[] | {kid, use, alg}'

B. Clock Skew Issues

# Issue: Token appears expired due to clock differences
# Solution: Add clock skew tolerance
decoded_token = jwt.decode(
    token,
    signing_key,
    algorithms=['RS256'],
    audience=audience,
    issuer=issuer,
    options={
        "verify_exp": True,
        "leeway": 60  # Allow 60 seconds clock skew
    }
)

C. Audience Mismatch

# Issue: "Invalid audience" error
# Debug: Check if audience claim matches your application
payload = jwt.decode(token, options={"verify_signature": False})
print(f"Token audience: {payload.get('aud')}")
print(f"Expected audience: {CLIENT_ID}")

JWT Roles in Hybrid Identity: MS Entra and Microsoft ADFS Federation

In enterprise environments where both cloud (MS Entra) and on-premises (Microsoft ADFS) identity providers coexist, JWT tokens play a crucial role in managing user roles and permissions across different systems. From my experience implementing hybrid identity solutions, role management becomes particularly complex when users need consistent access rights regardless of which identity provider authenticates them.

Understanding Role Challenges in Hybrid Environments

In hybrid identity scenarios, roles and permissions can be defined in multiple places:

• On-premises Active Directory (groups and roles)

• MS Entra ID (application roles and groups)

• ADFS claims (custom role mappings)

• Application-specific roles (local role definitions)

The challenge is ensuring consistent role representation in JWT tokens regardless of the authentication source.

JWT Role Structure in Hybrid Identity

MS Entra Role Claims in JWT

{
  "aud": "your-app-id",
  "iss": "https://login.microsoftonline.com/tenant-id/v2.0",
  "sub": "user-object-id",
  "preferred_username": "[email protected]",
  "roles": [
    "Application.Admin",
    "User.Read.All",
    "Reports.Manager"
  ],
  "groups": [
    "IT-Administrators",
    "Finance-Team",
    "Project-Managers"
  ],
  "wids": [
    "62e90394-69f5-4237-9190-012177145e10",  // Global Administrator
    "729827e3-9c14-49f7-bb1b-9608f156bbb8"   // Helpdesk Administrator
  ]
}

ADFS Role Claims in JWT

{
  "aud": "your-app-id",
  "iss": "https://adfs.company.local/adfs",
  "sub": "DOMAIN\\john.doe",
  "upn": "[email protected]",
  "http://schemas.microsoft.com/ws/2008/06/identity/claims/role": [
    "Domain Admins",
    "IT Support",
    "Finance Managers"
  ],
  "http://schemas.xmlsoap.org/claims/Group": [
    "CN=IT-Administrators,OU=Groups,DC=company,DC=local",
    "CN=Finance-Team,OU=Groups,DC=company,DC=local"
  ],
  "department": "Information Technology",
  "title": "Senior System Administrator"
}

Implementing Unified Role Management

Here's how I implement consistent role management across hybrid identity providers:

1. Role Mapping and Transformation

class HybridRoleManager:
    """Manage roles across hybrid identity providers"""
    
    def __init__(self):
        self.role_mappings = self._load_role_mappings()
        self.group_mappings = self._load_group_mappings()
    
    def _load_role_mappings(self) -> Dict[str, Dict[str, List[str]]]:
        """Load role mappings between different identity providers"""
        return {
            'entra_to_app': {
                'Application.Admin': ['admin', 'application_administrator'],
                'User.Read.All': ['user_reader', 'directory_reader'],
                'Reports.Manager': ['reports_admin', 'analytics_user']
            },
            'adfs_to_app': {
                'Domain Admins': ['admin', 'system_administrator'],
                'IT Support': ['support_user', 'helpdesk'],
                'Finance Managers': ['finance_admin', 'budget_manager']
            },
            'ad_groups_to_app': {
                'IT-Administrators': ['admin', 'it_admin'],
                'Finance-Team': ['finance_user', 'accounting'],
                'Project-Managers': ['project_admin', 'team_lead']
            }
        }
    
    def _load_group_mappings(self) -> Dict[str, List[str]]:
        """Load group to role mappings"""
        return {
            'IT-Administrators': ['admin', 'system_management', 'user_management'],
            'Finance-Team': ['finance_read', 'budget_view', 'expense_management'],
            'Project-Managers': ['project_admin', 'team_management', 'resource_allocation'],
            'Helpdesk': ['user_support', 'password_reset', 'basic_admin']
        }
    
    def extract_unified_roles(self, token: str, provider: str) -> List[str]:
        """Extract and unify roles from JWT token based on provider"""
        try:
            claims = jwt.decode(token, options={"verify_signature": False})
            
            if provider == 'entra':
                return self._extract_entra_roles(claims)
            elif provider == 'adfs':
                return self._extract_adfs_roles(claims)
            else:
                raise ValueError(f"Unsupported provider: {provider}")
                
        except Exception as e:
            print(f"Error extracting roles: {e}")
            return []
    
    def _extract_entra_roles(self, claims: Dict) -> List[str]:
        """Extract unified roles from MS Entra token"""
        unified_roles = set()
        
        # Extract application roles
        app_roles = claims.get('roles', [])
        for role in app_roles:
            mapped_roles = self.role_mappings['entra_to_app'].get(role, [])
            unified_roles.update(mapped_roles)
        
        # Extract group-based roles
        groups = claims.get('groups', [])
        for group in groups:
            mapped_roles = self.group_mappings.get(group, [])
            unified_roles.update(mapped_roles)
        
        # Extract directory roles (wids - Well-known IDs)
        wids = claims.get('wids', [])
        for wid in wids:
            if wid == '62e90394-69f5-4237-9190-012177145e10':  # Global Admin
                unified_roles.update(['admin', 'global_administrator'])
            elif wid == '729827e3-9c14-49f7-bb1b-9608f156bbb8':  # Helpdesk Admin
                unified_roles.update(['helpdesk', 'user_support'])
        
        return list(unified_roles)
    
    def _extract_adfs_roles(self, claims: Dict) -> List[str]:
        """Extract unified roles from ADFS token"""
        unified_roles = set()
        
        # Extract ADFS role claims
        role_claim_names = [
            'http://schemas.microsoft.com/ws/2008/06/identity/claims/role',
            'http://schemas.xmlsoap.org/claims/Role',
            'roles'
        ]
        
        for claim_name in role_claim_names:
            if claim_name in claims:
                roles = claims[claim_name]
                if isinstance(roles, str):
                    roles = [roles]
                
                for role in roles:
                    mapped_roles = self.role_mappings['adfs_to_app'].get(role, [])
                    unified_roles.update(mapped_roles)
        
        # Extract group-based roles
        group_claim_names = [
            'http://schemas.xmlsoap.org/claims/Group',
            'http://schemas.microsoft.com/ws/2008/06/identity/claims/groups',
            'groups'
        ]
        
        for claim_name in group_claim_names:
            if claim_name in claims:
                groups = claims[claim_name]
                if isinstance(groups, str):
                    groups = [groups]
                
                for group in groups:
                    # Extract group name from DN format
                    group_name = self._extract_group_name(group)
                    mapped_roles = self.group_mappings.get(group_name, [])
                    unified_roles.update(mapped_roles)
        
        return list(unified_roles)
    
    def _extract_group_name(self, group_dn: str) -> str:
        """Extract group name from LDAP Distinguished Name"""
        # Example: "CN=IT-Administrators,OU=Groups,DC=company,DC=local" -> "IT-Administrators"
        if group_dn.startswith('CN='):
            return group_dn.split(',')[0].replace('CN=', '')
        return group_dn
    
    def check_role_permission(self, user_roles: List[str], required_permission: str) -> bool:
        """Check if user roles include required permission"""
        permission_mappings = {
            'admin_access': ['admin', 'system_administrator', 'global_administrator'],
            'user_management': ['admin', 'user_admin', 'helpdesk'],
            'finance_access': ['finance_admin', 'finance_user', 'budget_manager'],
            'reports_access': ['reports_admin', 'analytics_user', 'admin'],
            'project_management': ['project_admin', 'team_lead', 'admin']
        }
        
        required_roles = permission_mappings.get(required_permission, [])
        return any(role in user_roles for role in required_roles)

2. Role-Based Authorization Middleware

class HybridRoleAuthMiddleware:
    """Middleware for role-based authorization in hybrid environments"""
    
    def __init__(self):
        self.role_manager = HybridRoleManager()
        self.token_validator = HybridTokenValidator()
    
    def authorize_request(self, request_headers: Dict, required_permission: str) -> Dict:
        """Authorize request based on JWT roles"""
        # Extract and validate token
        auth_result = self._validate_token(request_headers)
        if not auth_result['valid']:
            return {
                'authorized': False,
                'error': auth_result['error'],
                'status_code': 401
            }
        
        # Extract unified roles
        user_roles = self.role_manager.extract_unified_roles(
            auth_result['token'], 
            auth_result['provider']
        )
        
        # Check permission
        has_permission = self.role_manager.check_role_permission(
            user_roles, 
            required_permission
        )
        
        if has_permission:
            return {
                'authorized': True,
                'user_roles': user_roles,
                'provider': auth_result['provider'],
                'user_id': auth_result['user_id']
            }
        else:
            return {
                'authorized': False,
                'error': f'Insufficient permissions for {required_permission}',
                'user_roles': user_roles,
                'status_code': 403
            }
    
    def _validate_token(self, headers: Dict) -> Dict:
        """Validate JWT token and determine provider"""
        auth_header = headers.get('Authorization', '')
        if not auth_header.startswith('Bearer '):
            return {'valid': False, 'error': 'Missing authorization header'}
        
        token = auth_header.split(' ')[1]
        
        # Determine provider from token
        try:
            claims = jwt.decode(token, options={"verify_signature": False})
            issuer = claims.get('iss', '')
            
            if 'login.microsoftonline.com' in issuer:
                provider = 'entra'
            elif 'adfs' in issuer.lower():
                provider = 'adfs'
            else:
                return {'valid': False, 'error': 'Unknown token issuer'}
            
            # Validate token signature
            if self.token_validator.validate_token(token, provider):
                return {
                    'valid': True,
                    'token': token,
                    'provider': provider,
                    'user_id': claims.get('sub')
                }
            else:
                return {'valid': False, 'error': 'Invalid token signature'}
                
        except Exception as e:
            return {'valid': False, 'error': f'Token validation error: {str(e)}'}

3. Flask Application with Hybrid Role Authorization

from flask import Flask, request, jsonify
from functools import wraps

app = Flask(__name__)
auth_middleware = HybridRoleAuthMiddleware()

def require_permission(permission: str):
    """Decorator to require specific permission"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            auth_result = auth_middleware.authorize_request(
                request.headers.to_dict(), 
                permission
            )
            
            if not auth_result['authorized']:
                return jsonify({
                    'error': auth_result['error']
                }), auth_result.get('status_code', 403)
            
            # Add authorization info to request context
            request.auth_info = auth_result
            return f(*args, **kwargs)
        
        return decorated_function
    return decorator

@app.route('/admin/users', methods=['GET'])
@require_permission('user_management')
def list_users():
    """Endpoint requiring user management permission"""
    auth_info = request.auth_info
    
    return jsonify({
        'message': 'User list accessed successfully',
        'user_roles': auth_info['user_roles'],
        'provider': auth_info['provider'],
        'users': [
            {'id': 1, 'name': 'John Doe', 'email': '[email protected]'},
            {'id': 2, 'name': 'Jane Smith', 'email': '[email protected]'}
        ]
    })

@app.route('/finance/reports', methods=['GET'])
@require_permission('finance_access')
def get_finance_reports():
    """Endpoint requiring finance access permission"""
    auth_info = request.auth_info
    
    return jsonify({
        'message': 'Finance reports accessed successfully',
        'user_roles': auth_info['user_roles'],
        'provider': auth_info['provider'],
        'reports': [
            {'id': 1, 'name': 'Monthly Budget', 'type': 'budget'},
            {'id': 2, 'name': 'Expense Report', 'type': 'expense'}
        ]
    })

@app.route('/admin/system', methods=['GET'])
@require_permission('admin_access')
def system_admin():
    """Endpoint requiring admin access permission"""
    auth_info = request.auth_info
    
    return jsonify({
        'message': 'System admin area accessed successfully',
        'user_roles': auth_info['user_roles'],
        'provider': auth_info['provider'],
        'system_info': {
            'uptime': '15 days',
            'active_users': 1250,
            'system_health': 'healthy'
        }
    })

@app.route('/profile/roles', methods=['GET'])
def get_user_roles():
    """Get current user's roles and permissions"""
    auth_result = auth_middleware.authorize_request(
        request.headers.to_dict(), 
        'basic_access'  # Basic permission for authenticated users
    )
    
    if not auth_result['authorized']:
        return jsonify({
            'error': auth_result['error']
        }), auth_result.get('status_code', 401)
    
    # Get all permissions for user
    user_roles = auth_result['user_roles']
    permissions = []
    
    permission_checks = [
        'admin_access', 'user_management', 'finance_access', 
        'reports_access', 'project_management'
    ]
    
    for permission in permission_checks:
        if auth_middleware.role_manager.check_role_permission(user_roles, permission):
            permissions.append(permission)
    
    return jsonify({
        'user_id': auth_result['user_id'],
        'provider': auth_result['provider'],
        'roles': user_roles,
        'permissions': permissions
    })

if __name__ == '__main__':
    app.run(debug=True)

4. Role Synchronization Between Providers

class RoleSynchronizer:
    """Synchronize roles between MS Entra and ADFS"""
    
    def __init__(self):
        self.entra_client = MSGraphClient()
        self.adfs_client = ADFSClient()
        self.role_manager = HybridRoleManager()
    
    def sync_user_roles(self, user_identifier: str) -> Dict:
        """Synchronize user roles between providers"""
        try:
            # Get roles from both providers
            entra_roles = self._get_entra_user_roles(user_identifier)
            adfs_roles = self._get_adfs_user_roles(user_identifier)
            
            # Merge and normalize roles
            unified_roles = self._merge_roles(entra_roles, adfs_roles)
            
            # Update role cache
            self._update_role_cache(user_identifier, unified_roles)
            
            return {
                'success': True,
                'user_id': user_identifier,
                'entra_roles': entra_roles,
                'adfs_roles': adfs_roles,
                'unified_roles': unified_roles
            }
            
        except Exception as e:
            return {
                'success': False,
                'error': f'Role synchronization failed: {str(e)}'
            }
    
    def _get_entra_user_roles(self, user_id: str) -> List[str]:
        """Get user roles from MS Entra"""
        try:
            user_info = self.entra_client.get_user(user_id)
            app_roles = user_info.get('appRoleAssignments', [])
            groups = user_info.get('memberOf', [])
            
            roles = []
            for role in app_roles:
                roles.append(role.get('appRole', {}).get('displayName', ''))
            
            for group in groups:
                roles.append(group.get('displayName', ''))
            
            return [role for role in roles if role]  # Filter empty strings
            
        except Exception as e:
            print(f"Error getting Entra roles: {e}")
            return []
    
    def _get_adfs_user_roles(self, user_id: str) -> List[str]:
        """Get user roles from ADFS/AD"""
        try:
            user_info = self.adfs_client.get_user(user_id)
            groups = user_info.get('memberOf', [])
            
            roles = []
            for group_dn in groups:
                group_name = self.role_manager._extract_group_name(group_dn)
                roles.append(group_name)
            
            return roles
            
        except Exception as e:
            print(f"Error getting ADFS roles: {e}")
            return []
    
    def _merge_roles(self, entra_roles: List[str], adfs_roles: List[str]) -> List[str]:
        """Merge roles from different providers"""
        all_roles = set()
        
        # Map Entra roles
        for role in entra_roles:
            mapped_roles = self.role_manager.role_mappings['entra_to_app'].get(role, [])
            all_roles.update(mapped_roles)
        
        # Map ADFS roles
        for role in adfs_roles:
            mapped_roles = self.role_manager.role_mappings['adfs_to_app'].get(role, [])
            all_roles.update(mapped_roles)
        
        return list(all_roles)
    
    def _update_role_cache(self, user_id: str, roles: List[str]):
        """Update role cache for quick access"""
        # Implementation would depend on your caching strategy
        # Could be Redis, database, or in-memory cache
        pass

Best Practices for JWT Roles in Hybrid Identity

From my experience implementing role management in hybrid environments:

1. Consistent Role Mapping

• Create standardized role names across all providers

• Maintain mapping tables between provider-specific and application roles

• Use descriptive, business-meaningful role names

2. Hierarchical Role Structure

# Example role hierarchy
ROLE_HIERARCHY = {
    'admin': ['user', 'moderator', 'admin'],
    'moderator': ['user', 'moderator'],
    'user': ['user']
}

def check_role_hierarchy(user_roles: List[str], required_role: str) -> bool:
    """Check if user has required role or higher in hierarchy"""
    for user_role in user_roles:
        if required_role in ROLE_HIERARCHY.get(user_role, []):
            return True
    return False

3. Role Caching Strategy

• Cache role information to avoid repeated API calls

• Implement cache invalidation when roles change

• Set appropriate cache expiration times

4. Audit and Monitoring

• Log all role-based access decisions

• Monitor role assignments and changes

• Alert on unusual role access patterns

5. Security Considerations

• Validate role claims in JWT tokens

• Implement principle of least privilege

• Regular review and cleanup of role assignments

• Secure role synchronization processes

This comprehensive approach to JWT roles in hybrid identity environments ensures consistent, secure, and manageable role-based access control across different identity providers and applications.

Best Practices from My Experience

1. Security

• Always validate JWT signatures

• Use HTTPS for all communications

• Store tokens securely (avoid localStorage for sensitive apps)

• Implement proper token expiration handling

2. Performance

• Cache JWKS for signature validation

• Use connection pooling for HTTP requests

• Implement token caching with proper expiration

3. Error Handling

• Graceful handling of token expiration

• Proper error messages without exposing sensitive information

• Fallback mechanisms for authentication failures

4. Monitoring

• Log authentication events

• Monitor token usage patterns

• Alert on suspicious activities

Conclusion

JWT tokens with MS Entra provide a robust, scalable authentication solution. While they come with challenges like token size and revocation complexity, the benefits of stateless authentication and rich claim support make them ideal for modern applications.

The Python implementation demonstrated here provides a solid foundation for integrating JWT-based authentication in your applications. Remember to adapt the code to your specific security requirements and always follow the principle of least privilege when assigning roles and permissions.

In my experience, the key to successful JWT implementation is understanding the trade-offs and implementing proper security measures. MS Entra's robust infrastructure combined with careful implementation can provide enterprise-grade authentication for your applications.

Further Reading

• RFC 7519 - JSON Web Token (JWT)

• Microsoft Entra Documentation

• MSAL Python Documentation

• PyJWT Documentation