unionflow-server-api/diagnose_keycloak.py

#!/usr/bin/env python3
"""
Script de diagnostic pour Keycloak UnionFlow
"""

import requests
import json

class KeycloakDiagnostic:
    def __init__(self, base_url: str = "http://localhost:8180"):
        self.base_url = base_url
        self.session = requests.Session()
        self.admin_token = None
        
    def get_admin_token(self) -> bool:
        """Obtient le token admin"""
        try:
            data = {
                "username": "admin",
                "password": "admin",
                "grant_type": "password",
                "client_id": "admin-cli"
            }
            
            response = self.session.post(
                f"{self.base_url}/realms/master/protocol/openid-connect/token",
                data=data,
                headers={"Content-Type": "application/x-www-form-urlencoded"}
            )
            
            if response.status_code == 200:
                token_data = response.json()
                self.admin_token = token_data.get("access_token")
                return self.admin_token is not None
                
        except Exception as e:
            print(f"Erreur obtention token: {e}")
            
        return False
    
    def check_realm(self, realm_name: str = "unionflow") -> bool:
        """Vérifie le realm"""
        print(f"🔍 Vérification du realm {realm_name}...")
        
        try:
            response = self.session.get(
                f"{self.base_url}/admin/realms/{realm_name}",
                headers={"Authorization": f"Bearer {self.admin_token}"}
            )
            
            if response.status_code == 200:
                realm_data = response.json()
                print(f"✅ Realm {realm_name} existe")
                print(f"   - Enabled: {realm_data.get('enabled')}")
                print(f"   - Login with email: {realm_data.get('loginWithEmailAllowed')}")
                return True
            else:
                print(f"❌ Realm {realm_name} non trouvé: {response.status_code}")
                return False
                
        except Exception as e:
            print(f"❌ Erreur vérification realm: {e}")
            return False
    
    def check_client(self, realm_name: str = "unionflow", client_id: str = "unionflow-mobile") -> bool:
        """Vérifie le client"""
        print(f"🔍 Vérification du client {client_id}...")
        
        try:
            response = self.session.get(
                f"{self.base_url}/admin/realms/{realm_name}/clients",
                headers={"Authorization": f"Bearer {self.admin_token}"}
            )
            
            if response.status_code == 200:
                clients = response.json()
                for client in clients:
                    if client.get("clientId") == client_id:
                        print(f"✅ Client {client_id} trouvé")
                        print(f"   - Enabled: {client.get('enabled')}")
                        print(f"   - Public: {client.get('publicClient')}")
                        print(f"   - Direct Access Grants: {client.get('directAccessGrantsEnabled')}")
                        print(f"   - Standard Flow: {client.get('standardFlowEnabled')}")
                        return True
                
                print(f"❌ Client {client_id} non trouvé")
                print("Clients disponibles:")
                for client in clients:
                    print(f"   - {client.get('clientId')}")
                return False
            else:
                print(f"❌ Erreur récupération clients: {response.status_code}")
                return False
                
        except Exception as e:
            print(f"❌ Erreur vérification client: {e}")
            return False
    
    def check_users(self, realm_name: str = "unionflow") -> bool:
        """Vérifie les utilisateurs"""
        print(f"🔍 Vérification des utilisateurs...")
        
        expected_users = ["superadmin", "marie.active", "jean.simple", "tech.lead", "rh.manager"]
        
        try:
            response = self.session.get(
                f"{self.base_url}/admin/realms/{realm_name}/users",
                headers={"Authorization": f"Bearer {self.admin_token}"}
            )
            
            if response.status_code == 200:
                users = response.json()
                found_users = [user.get("username") for user in users]
                
                print(f"✅ {len(users)} utilisateurs trouvés")
                
                for expected in expected_users:
                    if expected in found_users:
                        print(f"   ✓ {expected}")
                    else:
                        print(f"   ✗ {expected} manquant")
                
                return len([u for u in expected_users if u in found_users]) == len(expected_users)
            else:
                print(f"❌ Erreur récupération utilisateurs: {response.status_code}")
                return False
                
        except Exception as e:
            print(f"❌ Erreur vérification utilisateurs: {e}")
            return False
    
    def test_direct_auth(self, realm_name: str = "unionflow", client_id: str = "unionflow-mobile"):
        """Teste l'authentification directe"""
        print(f"🔍 Test d'authentification directe...")
        
        # Test avec marie.active
        try:
            data = {
                "username": "marie.active",
                "password": "Marie123!",
                "grant_type": "password",
                "client_id": client_id
            }
            
            response = self.session.post(
                f"{self.base_url}/realms/{realm_name}/protocol/openid-connect/token",
                data=data,
                headers={"Content-Type": "application/x-www-form-urlencoded"}
            )
            
            print(f"Status: {response.status_code}")
            print(f"Response: {response.text[:200]}...")
            
            if response.status_code == 200:
                print("✅ Authentification réussie")
            else:
                print("❌ Authentification échouée")
                
        except Exception as e:
            print(f"❌ Erreur test auth: {e}")
    
    def diagnose(self):
        """Lance le diagnostic complet"""
        print("=" * 80)
        print("🔍 DIAGNOSTIC KEYCLOAK UNIONFLOW")
        print("=" * 80)
        print()
        
        # 1. Vérifier connexion Keycloak
        try:
            response = self.session.get(f"{self.base_url}", timeout=5)
            if response.status_code == 200:
                print("✅ Keycloak accessible")
            else:
                print(f"❌ Keycloak non accessible: {response.status_code}")
                return
        except:
            print("❌ Keycloak non accessible")
            return
        
        print()
        
        # 2. Obtenir token admin
        if not self.get_admin_token():
            print("❌ Impossible d'obtenir le token admin")
            return
        
        print("✅ Token admin obtenu")
        print()
        
        # 3. Vérifier realm
        self.check_realm()
        print()
        
        # 4. Vérifier client
        self.check_client()
        print()
        
        # 5. Vérifier utilisateurs
        self.check_users()
        print()
        
        # 6. Test authentification
        self.test_direct_auth()
        print()
        
        print("=" * 80)
        print("🔍 DIAGNOSTIC TERMINÉ")
        print("=" * 80)


def main():
    diagnostic = KeycloakDiagnostic()
    diagnostic.diagnose()


if __name__ == "__main__":
    main()