JWT Auditor MCP Server

Instructions

Implementation Reference

• server.py:31-91 (handler)

  Implementation of the 'jwt_analyze' tool handler. This function decodes the JWT, checks for various vulnerabilities like none alg, weak alg, sensitive data, missing claims, header injection, expiration, replay risks, and alg confusion. Returns findings list along with header and payload.
  @server.tool() def jwt_analyze(token: str) -> dict: """Analyze a JWT for common vulnerabilities and issues.""" import re import time findings = [] try: header_b64, payload_b64, signature_b64 = token.split(".") def b64decode(data): rem = len(data) % 4 if rem: data += '=' * (4 - rem) return base64.urlsafe_b64decode(data.encode()) header = json.loads(b64decode(header_b64)) payload = json.loads(b64decode(payload_b64)) # 1. Algorithm vulnerabilities alg = header.get("alg", "") if alg.lower() == "none": findings.append({"type": "alg_none", "severity": "critical", "description": "JWT uses 'none' algorithm (no signature)."}) if alg.lower() in ["hs256", "hs384", "hs512"]: findings.append({"type": "weak_alg", "severity": "warning", "description": f"JWT uses symmetric algorithm: {alg}. Check for secret reuse and brute-force risk."}) if alg.lower() in ["rs256", "rs384", "rs512"]: findings.append({"type": "asymmetric_alg", "severity": "info", "description": f"JWT uses asymmetric algorithm: {alg}."}) # 2. Sensitive data exposure sensitive_patterns = [ (re.compile(r"(password|secret|api[_-]?key|access[_-]?token)", re.I), "Potential credential exposure"), (re.compile(r"\b(?:\d[ -]*?){13,16}\b"), "Possible credit card number"), (re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}"), "Possible email address (PII)") ] for k, v in payload.items(): for pat, desc in sensitive_patterns: if isinstance(v, str) and pat.search(v): findings.append({"type": "sensitive_data", "severity": "warning", "description": f"{desc} in claim '{k}'"}) # 3. Missing security claims for claim in ["exp", "iss", "aud", "jti"]: if claim not in payload: findings.append({"type": "missing_claim", "severity": "warning", "description": f"Missing recommended claim: {claim}"}) # 4. Header injection vulnerabilities (kid) if "kid" in header: if isinstance(header["kid"], str) and (".." in header["kid"] or "/" in header["kid"]): findings.append({"type": "header_injection", "severity": "critical", "description": "Potential header injection in 'kid' parameter."}) # 5. Token lifetime and replay attack analysis now = int(time.time()) if "exp" in payload: try: exp = int(payload["exp"]) if exp < now: findings.append({"type": "expired", "severity": "critical", "description": "Token is expired."}) elif exp - now > 60*60*24*7: findings.append({"type": "long_lifetime", "severity": "warning", "description": "Token lifetime is unusually long (> 7 days)."}) except Exception: findings.append({"type": "invalid_exp", "severity": "warning", "description": "Invalid 'exp' claim format."}) # 6. Replay attack analysis (jti) if "jti" not in payload: findings.append({"type": "replay_attack", "severity": "info", "description": "No 'jti' claim: token may be replayable."}) # 7. Algorithm confusion (alg confusion attacks) if alg.lower().startswith("hs") and "kid" in header and header["kid"].endswith(".pem"): findings.append({"type": "alg_confusion", "severity": "critical", "description": "Potential algorithm confusion: HS* with 'kid' referencing a PEM file."}) return {"findings": findings, "header": header, "payload": payload} except Exception as e: return {"error": str(e)}