#!/usr/bin/env python3 """ Update Keycloak user club attributes to use real database UUIDs instead of placeholders. This script updates the 'clubs' attribute for all test users in the workclub realm, replacing placeholder UUIDs with real database UUIDs and restarting Keycloak to ensure tokens reflect the updated attributes. Real UUIDs: - Sunrise Tennis Club: afa8daf3-5cfa-4589-9200-b39a538a12de - Valley Cycling Club: a1952a72-2e13-4a4e-87dd-821847b58698 Usage: python3 update-keycloak-club-uuids.py """ import json import subprocess import sys KEYCLOAK_ADMIN_URL = "http://localhost:8080" KEYCLOAK_REALM = "workclub" DB_HOST = "localhost" DB_PORT = 5432 DB_NAME = "keycloak" DB_USER = "postgres" SUNRISE_UUID = "afa8daf3-5cfa-4589-9200-b39a538a12de" VALLEY_UUID = "a1952a72-2e13-4a4e-87dd-821847b58698" # User ID mappings from Keycloak USERS = { "bf5adcfb-0978-4beb-8e02-7577f0ded47f": { "username": "admin@test.com", "clubs": {SUNRISE_UUID: "admin", VALLEY_UUID: "member"} }, "aa5270a3-633a-4d89-a3b4-a467b08cbb55": { "username": "manager@test.com", "clubs": {SUNRISE_UUID: "manager", VALLEY_UUID: "member"} }, "60c0d8b9-6354-4ad3-bfac-9547c68c069b": { "username": "member1@test.com", "clubs": {SUNRISE_UUID: "member", VALLEY_UUID: "member"} }, "294a2086-cf2f-43cc-9bc6-2a8a7d325b9a": { "username": "member2@test.com", "clubs": {VALLEY_UUID: "member"} }, "f4890d47-ba6c-4691-9d7b-4f656c60f232": { "username": "viewer@test.com", "clubs": {SUNRISE_UUID: "viewer"} } } def update_via_database(): """Update clubs attributes directly in PostgreSQL database.""" print("\n[1/4] Updating clubs attributes in database...") for user_id, user_data in USERS.items(): username = user_data["username"] clubs = user_data["clubs"] clubs_json = json.dumps(clubs).replace("'", "''") sql = f"UPDATE user_attribute SET value = '{clubs_json}' WHERE user_id = '{user_id}' AND name = 'clubs';" cmd = [ "docker", "exec", "workclub_postgres", "psql", "-U", DB_USER, "-d", DB_NAME, "-c", sql ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print(f" ✗ {username}: {result.stderr}") return False print(f" ✓ {username}") return True def restart_keycloak(): """Restart Keycloak to clear caches.""" print("\n[2/4] Restarting Keycloak to clear caches...") cmd = ["docker", "restart", "workclub_keycloak"] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print(f" ✗ Failed to restart: {result.stderr}") return False print(" ✓ Keycloak restarted") # Wait for Keycloak to be ready print(" ⏳ Waiting for Keycloak to be ready...") for i in range(30): try: cmd = [ "curl", "-s", "-f", f"{KEYCLOAK_ADMIN_URL}/health" ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=2) if result.returncode == 0: print(" ✓ Keycloak is ready") return True except: pass if (i + 1) % 5 == 0: print(f" Waiting... ({i + 1}/30)") print(" ✗ Keycloak did not become ready in time") return False def verify_jwt_tokens(): """Verify JWT tokens contain real UUIDs.""" import base64 print("\n[3/4] Verifying JWT tokens contain real UUIDs...") for user_id, user_data in USERS.items(): username = user_data["username"] # Get token cmd = [ "curl", "-s", "-X", "POST", f"{KEYCLOAK_ADMIN_URL}/realms/{KEYCLOAK_REALM}/protocol/openid-connect/token", "-d", "client_id=workclub-app", "-d", "grant_type=password", "-d", f"username={username}", "-d", "password=testpass123" ] result = subprocess.run(cmd, capture_output=True, text=True) try: data = json.loads(result.stdout) if "access_token" not in data: print(f" ✗ {username}: {data.get('error', 'No token')}") return False token = data["access_token"] parts = token.split('.') # Decode payload payload = parts[1] padding = 4 - len(payload) % 4 if padding != 4: payload += '=' * padding decoded = json.loads(base64.urlsafe_b64decode(payload)) clubs = decoded.get("clubs", {}) # Check if real UUIDs are present expected_clubs = user_data["clubs"] if clubs == expected_clubs: print(f" ✓ {username}: {json.dumps(clubs)}") else: print(f" ✗ {username}: Expected {expected_clubs}, got {clubs}") return False except Exception as e: print(f" ✗ {username}: {str(e)}") return False return True def test_api_endpoint(): """Test API endpoint with real UUIDs.""" print("\n[4/4] Testing API endpoints with real UUIDs...") # Test with admin user and first club cmd = [ "curl", "-s", "-X", "POST", f"{KEYCLOAK_ADMIN_URL}/realms/{KEYCLOAK_REALM}/protocol/openid-connect/token", "-d", "client_id=workclub-app", "-d", "grant_type=password", "-d", "username=admin@test.com", "-d", "password=testpass123" ] result = subprocess.run(cmd, capture_output=True, text=True) token = json.loads(result.stdout)["access_token"] # Test API cmd = [ "curl", "-s", "-i", "-X", "GET", "-H", f"Authorization: Bearer {token}", "-H", f"X-Tenant-Id: {SUNRISE_UUID}", "http://localhost:5001/api/clubs/me" ] result = subprocess.run(cmd, capture_output=True, text=True) if "200 OK" in result.stdout: print(f" ✓ GET /api/clubs/me: 200 OK with real UUID in X-Tenant-Id") return True else: print(f" ✗ API endpoint returned: {result.stdout[:100]}") return False def main(): """Run all update steps.""" print("=" * 80) print("KEYCLOAK CLUB UUID UPDATE SCRIPT") print("=" * 80) steps = [ ("Update database", update_via_database), ("Restart Keycloak", restart_keycloak), ("Verify JWT tokens", verify_jwt_tokens), ("Test API endpoint", test_api_endpoint), ] for step_name, step_func in steps: if not step_func(): print(f"\n✗ FAILED at: {step_name}") return 1 print("\n" + "=" * 80) print("✓ ALL UPDATES SUCCESSFUL") print("=" * 80) print("\nSummary:") print(f" - Updated 5 test users with real club UUIDs") print(f" - Sunrise Tennis Club: {SUNRISE_UUID}") print(f" - Valley Cycling Club: {VALLEY_UUID}") print(f" - JWT tokens now contain real UUIDs instead of placeholders") print(f" - API endpoints accept requests with real X-Tenant-Id headers") return 0 if __name__ == "__main__": sys.exit(main())