unionflow-server-api/unionflow/scripts/keycloak-setup.py

#!/usr/bin/env python3
"""
Script de configuration Keycloak pour UnionFlow
Crée les rôles et les comptes de test pour MUKEFI et MESKA
"""

import requests
import json
import sys
from typing import Dict, List, Optional

# Configuration
KEYCLOAK_URL = "http://localhost:8180"
REALM_NAME = "unionflow"
ADMIN_USERNAME = "admin"
ADMIN_PASSWORD = "admin"  # Modifier si différent
TEST_PASSWORD = "Test@123"

class KeycloakManager:
    def __init__(self):
        self.base_url = KEYCLOAK_URL
        self.realm = REALM_NAME
        self.token = None

    def get_admin_token(self) -> bool:
        """Obtient un token admin pour l'API Keycloak"""
        url = f"{self.base_url}/realms/master/protocol/openid-connect/token"
        data = {
            "client_id": "admin-cli",
            "username": ADMIN_USERNAME,
            "password": ADMIN_PASSWORD,
            "grant_type": "password"
        }

        try:
            response = requests.post(url, data=data)
            response.raise_for_status()
            self.token = response.json()["access_token"]
            print("✅ Connecté à Keycloak Admin API")
            return True
        except Exception as e:
            print(f"❌ Erreur connexion Keycloak: {e}")
            return False

    def _headers(self) -> Dict:
        """Headers pour les requêtes API"""
        return {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }

    def list_users(self) -> List[Dict]:
        """Liste tous les utilisateurs du realm"""
        url = f"{self.base_url}/admin/realms/{self.realm}/users"
        response = requests.get(url, headers=self._headers())
        response.raise_for_status()
        return response.json()

    def delete_user(self, user_id: str) -> bool:
        """Supprime un utilisateur"""
        url = f"{self.base_url}/admin/realms/{self.realm}/users/{user_id}"
        response = requests.delete(url, headers=self._headers())
        return response.status_code == 204

    def list_roles(self) -> List[Dict]:
        """Liste tous les rôles du realm"""
        url = f"{self.base_url}/admin/realms/{self.realm}/roles"
        response = requests.get(url, headers=self._headers())
        response.raise_for_status()
        return response.json()

    def create_role(self, name: str, description: str) -> bool:
        """Crée un rôle realm"""
        url = f"{self.base_url}/admin/realms/{self.realm}/roles"
        data = {
            "name": name,
            "description": description,
            "composite": False,
            "clientRole": False
        }

        try:
            response = requests.post(url, headers=self._headers(), json=data)
            if response.status_code == 201:
                print(f"  ✅ Rôle créé: {name}")
                return True
            elif response.status_code == 409:
                print(f"  ⚠️  Rôle existe déjà: {name}")
                return True
            else:
                print(f"  ❌ Erreur création rôle {name}: {response.status_code}")
                return False
        except Exception as e:
            print(f"  ❌ Exception création rôle {name}: {e}")
            return False

    def create_user(self, username: str, email: str, first_name: str, last_name: str,
                   password: str, roles: List[str], enabled: bool = True) -> Optional[str]:
        """Crée un utilisateur avec ses rôles"""
        # 1. Créer l'utilisateur
        url = f"{self.base_url}/admin/realms/{self.realm}/users"
        data = {
            "username": username,
            "email": email,
            "firstName": first_name,
            "lastName": last_name,
            "enabled": enabled,
            "emailVerified": True,
            "credentials": [{
                "type": "password",
                "value": password,
                "temporary": False
            }]
        }

        try:
            response = requests.post(url, headers=self._headers(), json=data)

            if response.status_code == 201:
                # Récupérer l'ID de l'utilisateur créé
                location = response.headers.get("Location")
                user_id = location.split("/")[-1] if location else None

                if not user_id:
                    # Chercher l'utilisateur par username
                    users = self.list_users()
                    user = next((u for u in users if u["username"] == username), None)
                    if user:
                        user_id = user["id"]

                if user_id:
                    # 2. Assigner les rôles
                    self.assign_roles_to_user(user_id, roles)
                    print(f"  ✅ Utilisateur créé: {username} ({email})")
                    return user_id
                else:
                    print(f"  ⚠️  Utilisateur créé mais ID non trouvé: {username}")
                    return None
            elif response.status_code == 409:
                print(f"  ⚠️  Utilisateur existe déjà: {username}")
                # Récupérer l'ID et mettre à jour les rôles
                users = self.list_users()
                user = next((u for u in users if u["username"] == username), None)
                if user:
                    self.assign_roles_to_user(user["id"], roles)
                return None
            else:
                print(f"  ❌ Erreur création utilisateur {username}: {response.status_code} - {response.text}")
                return None
        except Exception as e:
            print(f"  ❌ Exception création utilisateur {username}: {e}")
            return None

    def assign_roles_to_user(self, user_id: str, role_names: List[str]):
        """Assigne des rôles à un utilisateur"""
        # Récupérer les objets de rôle
        all_roles = self.list_roles()
        roles_to_assign = [r for r in all_roles if r["name"] in role_names]

        if not roles_to_assign:
            return

        url = f"{self.base_url}/admin/realms/{self.realm}/users/{user_id}/role-mappings/realm"
        response = requests.post(url, headers=self._headers(), json=roles_to_assign)

        if response.status_code in [204, 200]:
            print(f"    → Rôles assignés: {', '.join(role_names)}")
        else:
            print(f"    ⚠️  Erreur assignation rôles: {response.status_code}")

def main():
    print("=" * 70)
    print("🔧 Configuration Keycloak pour UnionFlow")
    print("=" * 70)

    kc = KeycloakManager()

    # 1. Connexion
    print("\n📡 Connexion à Keycloak...")
    if not kc.get_admin_token():
        print("\n❌ Impossible de se connecter à Keycloak.")
        print("   Vérifiez que Keycloak est démarré sur http://localhost:8180")
        print("   Credentials par défaut: admin / admin")
        sys.exit(1)

    # 2. Audit de l'existant
    print("\n📋 Audit de l'état actuel...")
    existing_users = kc.list_users()
    existing_roles = kc.list_roles()

    print(f"  • Utilisateurs existants: {len(existing_users)}")
    for user in existing_users:
        print(f"    - {user['username']} ({user.get('email', 'no email')})")

    print(f"  • Rôles existants: {len(existing_roles)}")
    for role in existing_roles:
        print(f"    - {role['name']}")

    # 3. Suppression des utilisateurs existants
    print("\n🗑️  Suppression des utilisateurs existants...")
    for user in existing_users:
        if kc.delete_user(user["id"]):
            print(f"  ✅ Supprimé: {user['username']}")
        else:
            print(f"  ❌ Erreur suppression: {user['username']}")

    # 4. Création de la structure de rôles
    print("\n👥 Création de la structure de rôles...")

    roles_to_create = [
        ("SUPER_ADMIN", "Super administrateur - Accès total plateforme multi-organisations"),
        ("ADMIN_ORGANISATION", "Administrateur d'une organisation - Accès total à son organisation"),
        ("TRESORIER", "Trésorier - Gestion financière, comptabilité, épargne/crédit"),
        ("SECRETAIRE", "Secrétaire - Gestion administrative, membres, adhésions, documents"),
        ("RESPONSABLE_SOCIAL", "Responsable social - Gestion aide sociale et solidarité"),
        ("RESPONSABLE_EVENEMENTS", "Responsable événements - Gestion événements et logistique"),
        ("RESPONSABLE_CREDIT", "Responsable crédit - Gestion épargne/crédit (mutuelles)"),
        ("MEMBRE_BUREAU", "Membre du bureau - Accès étendu consultation et actions"),
        ("MEMBRE_ACTIF", "Membre actif - Consultation et actions de base"),
        ("MEMBRE_SIMPLE", "Membre simple - Consultation uniquement"),
        ("MEMBRE", "Rôle technique - Membre base"),
        ("ADMIN", "Rôle technique - Admin base"),
        ("USER", "Rôle technique - Utilisateur base")
    ]

    for role_name, description in roles_to_create:
        kc.create_role(role_name, description)

    # 5. Création des comptes de test
    print("\n👤 Création des comptes de test...")

    users_to_create = [
        # Super-Admin
        {
            "username": "superadmin",
            "email": "superadmin@unionflow.test",
            "first_name": "Super",
            "last_name": "Admin",
            "roles": ["SUPER_ADMIN", "ADMIN", "USER"]
        },

        # MUKEFI (Mutuelle d'épargne et de crédit)
        {
            "username": "admin.mukefi",
            "email": "admin.mukefi@unionflow.test",
            "first_name": "Administrateur",
            "last_name": "MUKEFI",
            "roles": ["ADMIN_ORGANISATION", "ADMIN", "USER"]
        },
        {
            "username": "tresorier.mukefi",
            "email": "tresorier.mukefi@unionflow.test",
            "first_name": "Trésorier",
            "last_name": "MUKEFI",
            "roles": ["TRESORIER", "MEMBRE", "USER"]
        },
        {
            "username": "secretaire.mukefi",
            "email": "secretaire.mukefi@unionflow.test",
            "first_name": "Secrétaire",
            "last_name": "MUKEFI",
            "roles": ["SECRETAIRE", "MEMBRE", "USER"]
        },
        {
            "username": "credit.mukefi",
            "email": "credit.mukefi@unionflow.test",
            "first_name": "Responsable Crédit",
            "last_name": "MUKEFI",
            "roles": ["RESPONSABLE_CREDIT", "MEMBRE", "USER"]
        },
        {
            "username": "membre.mukefi",
            "email": "membre.mukefi@unionflow.test",
            "first_name": "Membre",
            "last_name": "MUKEFI",
            "roles": ["MEMBRE_ACTIF", "MEMBRE", "USER"]
        },

        # MESKA (Association)
        {
            "username": "admin.meska",
            "email": "admin.meska@unionflow.test",
            "first_name": "Administrateur",
            "last_name": "MESKA",
            "roles": ["ADMIN_ORGANISATION", "ADMIN", "USER"]
        },
        {
            "username": "secretaire.meska",
            "email": "secretaire.meska@unionflow.test",
            "first_name": "Secrétaire",
            "last_name": "MESKA",
            "roles": ["SECRETAIRE", "MEMBRE", "USER"]
        },
        {
            "username": "social.meska",
            "email": "social.meska@unionflow.test",
            "first_name": "Responsable Social",
            "last_name": "MESKA",
            "roles": ["RESPONSABLE_SOCIAL", "MEMBRE", "USER"]
        },
        {
            "username": "evenements.meska",
            "email": "evenements.meska@unionflow.test",
            "first_name": "Responsable Événements",
            "last_name": "MESKA",
            "roles": ["RESPONSABLE_EVENEMENTS", "MEMBRE", "USER"]
        },
        {
            "username": "membre.meska",
            "email": "membre.meska@unionflow.test",
            "first_name": "Membre",
            "last_name": "MESKA",
            "roles": ["MEMBRE_ACTIF", "MEMBRE", "USER"]
        }
    ]

    print(f"\n📝 Création de {len(users_to_create)} comptes utilisateurs...")
    print(f"   Mot de passe pour tous: {TEST_PASSWORD}\n")

    for user_data in users_to_create:
        kc.create_user(
            username=user_data["username"],
            email=user_data["email"],
            first_name=user_data["first_name"],
            last_name=user_data["last_name"],
            password=TEST_PASSWORD,
            roles=user_data["roles"]
        )

    # 6. Résumé final
    print("\n" + "=" * 70)
    print("✅ Configuration Keycloak terminée avec succès !")
    print("=" * 70)

    print("\n📊 Résumé:")
    print(f"  • {len(roles_to_create)} rôles créés")
    print(f"  • {len(users_to_create)} utilisateurs créés")
    print(f"  • Mot de passe: {TEST_PASSWORD}")

    print("\n👥 Comptes créés:")
    print("\n  🔧 Super-Admin:")
    print("    → superadmin@unionflow.test")

    print("\n  🏦 MUKEFI (Mutuelle):")
    print("    → admin.mukefi@unionflow.test (Admin)")
    print("    → tresorier.mukefi@unionflow.test (Trésorier)")
    print("    → secretaire.mukefi@unionflow.test (Secrétaire)")
    print("    → credit.mukefi@unionflow.test (Responsable Crédit)")
    print("    → membre.mukefi@unionflow.test (Membre Actif)")

    print("\n  🤝 MESKA (Association):")
    print("    → admin.meska@unionflow.test (Admin)")
    print("    → secretaire.meska@unionflow.test (Secrétaire)")
    print("    → social.meska@unionflow.test (Responsable Social)")
    print("    → evenements.meska@unionflow.test (Responsable Événements)")
    print("    → membre.meska@unionflow.test (Membre Actif)")

    print("\n🌐 Accès Keycloak:")
    print(f"  • Console Admin: {KEYCLOAK_URL}/admin")
    print(f"  • Realm: {REALM_NAME}")

    print("\n")

if __name__ == "__main__":
    main()