Patched code sample

# This code demonstrates the patched behavior of PyJWT (version 2.13.0 and later,
# though the specific fix for this type of issue was introduced earlier).
# The PyJWKClient now caches the JWKS response, preventing repeated requests for
# different unknown 'kid' values.

import http.server
import json
import socketserver
import threading
import time

# PyJWT is required. Install it with: pip install "PyJWT[crypto]"
from jwt import PyJWKClient
from jwt.exceptions import PyJWKClientError

# --- 1. Set up a mock JWKS endpoint ---
# This server will log every time it receives a request, allowing us to see
# how many times the PyJWKClient actually hits the network.

PORT = 8000
JWKS_URI = f"http://localhost:{PORT}/.well-known/jwks.json"

# A valid JWKS with one key. The demonstration will use kids that are NOT this one.
JWKS_CONTENT = {
    "keys": [
        {
            "kty": "RSA",
            "kid": "valid-key-id",
            "use": "sig",
            "n": "...",
            "e": "AQAB",
        }
    ]
}


class JwksHttpRequestHandler(http.server.SimpleHTTPRequestHandler):
    def do_GET(self):
        # This print statement is the core of the demonstration.
        # It will only appear when a real HTTP request is made.
        print(f"--- JWKS endpoint was requested by the client at {time.time():.2f} ---")
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(JWKS_CONTENT).encode("utf-8"))


def run_server():
    with socketserver.TCPServer(("", PORT), JwksHttpRequestHandler) as httpd:
        # print(f"Mock JWKS server running at port {PORT}")
        httpd.serve_forever()


# Run the server in a background thread
server_thread = threading.Thread(target=run_server)
server_thread.daemon = True
server_thread.start()
time.sleep(0.5)  # Give the server a moment to start up

# --- 2. Demonstrate the fixed (caching) behavior ---

# Initialize the client with a short cache lifespan for demonstration purposes.
# The client will cache the JWKS content for 5 seconds.
jwk_client = PyJWKClient(JWKS_URI, lifespan=5)

print("Attempting to get 3 different unknown keys in quick succession...")

for i in range(3):
    unknown_kid = f"attacker-kid-{i}"
    try:
        print(f"Client: Requesting key with kid: '{unknown_kid}'")
        jwk_client.get_signing_key(kid=unknown_kid)
    except PyJWKClientError as e:
        print(f"Client: Correctly failed to find key '{unknown_kid}'.")

# In the vulnerable version, the server would have received 3 requests.
# In the fixed version, it receives only one request because the result
# (the full set of keys) is cached.

print("\nWaiting for cache to expire (lifespan is 5 seconds)...")
time.sleep(6)

print("\nAttempting to get another unknown key after cache expiry...")
try:
    print("Client: Requesting key with kid: 'another-attacker-kid'")
    jwk_client.get_signing_key(kid="another-attacker-kid")
except PyJWKClientError as e:
    print("Client: Correctly failed to find key 'another-attacker-kid'.")

# Because the cache expired, the client makes a second network request to
# refresh its cache before failing to find the key.

print("\nDemonstration complete. The server was only requested twice.")
# To stop the script, we would normally shut down the server, but since
# it's a daemon thread, it will exit when the main script finishes.