From 3b5c183b47d2ddbdfad75df759e822d49eaf8b13 Mon Sep 17 00:00:00 2001 From: Fanilo-Nantenaina Date: Wed, 21 Jan 2026 16:53:38 +0300 Subject: [PATCH] refactor(security): remove emojis from logs and improve script debugging --- middleware/security.py | 6 +- scripts/manage_security.py | 261 +++++++++++++++++++++++++++++++++---- 2 files changed, 236 insertions(+), 31 deletions(-) diff --git a/middleware/security.py b/middleware/security.py index 0616595..9501d5b 100644 --- a/middleware/security.py +++ b/middleware/security.py @@ -168,7 +168,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware): if token.startswith("sdk_live_"): logger.warning( - "⚠️ API Key envoyée dans Authorization au lieu de X-API-Key" + " API Key envoyée dans Authorization au lieu de X-API-Key" ) return await self._handle_api_key_auth( request, token, path, method, call_next @@ -184,7 +184,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware): request, api_key_header, path, method, call_next ) - logger.debug(f"❌ Aucune auth pour {method} {path} → délégation à FastAPI") + logger.debug(f" Aucune auth pour {method} {path} → délégation à FastAPI") return await call_next(request) async def _handle_api_key_auth( @@ -255,7 +255,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware): request.state.api_key = api_key_obj request.state.authenticated_via = "api_key" - logger.info(f"✅ ACCÈS AUTORISÉ: {api_key_obj.name} → {method} {path}") + logger.info(f" ACCÈS AUTORISÉ: {api_key_obj.name} → {method} {path}") return await call_next(request) diff --git a/scripts/manage_security.py b/scripts/manage_security.py index c8ff0bd..6e03ef5 100644 --- a/scripts/manage_security.py +++ b/scripts/manage_security.py @@ -1,9 +1,3 @@ -#!/usr/bin/env python3 -""" -Script de gestion avancée des utilisateurs Swagger et API Keys -avec configuration des schémas d'authentification -""" - import sys import os from pathlib import Path @@ -19,12 +13,35 @@ _current_file = Path(__file__).resolve() _script_dir = _current_file.parent _app_dir = _script_dir.parent +print(f"DEBUG: Script path: {_current_file}") +print(f"DEBUG: App dir: {_app_dir}") +print(f"DEBUG: Current working dir: {os.getcwd()}") + if str(_app_dir) in sys.path: sys.path.remove(str(_app_dir)) sys.path.insert(0, str(_app_dir)) os.chdir(str(_app_dir)) +print(f"DEBUG: sys.path[0]: {sys.path[0]}") +print(f"DEBUG: New working dir: {os.getcwd()}") + +_test_imports = [ + "database", + "database.db_config", + "database.models", + "services", + "security", +] + +print("\nDEBUG: Vérification des imports...") +for module in _test_imports: + try: + __import__(module) + print(f" ✓ {module}") + except ImportError as e: + print(f" ✗ {module}: {e}") + try: from database.db_config import async_session_factory from database.models.api_key import SwaggerUser, ApiKey @@ -33,6 +50,7 @@ try: except ImportError as e: print(f"\n ERREUR D'IMPORT: {e}") print(" Vérifiez que vous êtes dans /app") + print(" Commande correcte: cd /app && python scripts/manage_security.py ...") sys.exit(1) logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s") @@ -40,8 +58,8 @@ logger = logging.getLogger(__name__) AVAILABLE_TAGS = { - "Authentication": "🔐 Authentification et gestion des comptes", - "API Keys Management": " Gestion des clés API", + "Authentication": " Authentification et gestion des comptes", + "API Keys Management": "🔑 Gestion des clés API", "Clients": "👥 Gestion des clients", "Fournisseurs": "🏭 Gestion des fournisseurs", "Prospects": "🎯 Gestion des prospects", @@ -56,13 +74,13 @@ AVAILABLE_TAGS = { "Factures": "💰 Factures", "Avoirs": "↩️ Avoirs", "Règlements": "💳 Règlements et encaissements", - "Workflows": "🔄 Transformations de documents", + "Workflows": " Transformations de documents", "Documents": "📑 Gestion documents (PDF)", "Emails": "📧 Envoi d'emails", "Validation": " Validations métier", "Collaborateurs": "👔 Collaborateurs internes", "Société": "🏢 Informations société", - "Référentiels": " Données de référence", + "Référentiels": "📚 Données de référence", "System": "⚙️ Système et santé", "Admin": "🛠️ Administration", "Debug": "🐛 Debug et diagnostics", @@ -199,16 +217,16 @@ async def list_swagger_users(): auth_schemes.append("JWT (Bearer)") logger.info( - f"🔐 Authentification autorisée: {', '.join(auth_schemes)}" + f" Authentification autorisée: {', '.join(auth_schemes)}" ) else: logger.info("👑 Tags autorisés: ADMIN COMPLET (tous)") - logger.info("🔐 Authentification: JWT + X-API-Key (tout)") + logger.info(" Authentification: JWT + X-API-Key (tout)") except json.JSONDecodeError: - logger.info("⚠️ Tags: Erreur format") + logger.info(" Tags: Erreur format") else: logger.info("👑 Tags autorisés: ADMIN COMPLET (tous)") - logger.info("🔐 Authentification: JWT + X-API-Key (tout)") + logger.info(" Authentification: JWT + X-API-Key (tout)") logger.info("\n" + "=" * 80) @@ -244,7 +262,7 @@ async def update_swagger_user( elif set_tags is not None: user.allowed_tags = json.dumps(set_tags) if set_tags else None - logger.info(f"🔄 Tags remplacés: {len(set_tags) if set_tags else 0}") + logger.info(f" Tags remplacés: {len(set_tags) if set_tags else 0}") modified = True elif add_tags or remove_tags: @@ -273,19 +291,36 @@ async def update_swagger_user( if active is not None: user.is_active = active - logger.info(f"🔄 Statut: {'ACTIF' if active else 'INACTIF'}") + logger.info(f" Statut: {'ACTIF' if active else 'INACTIF'}") modified = True if modified: await session.commit() logger.info(f" Utilisateur '{username}' mis à jour") else: - logger.info("ℹ️ Aucune modification effectuée") + logger.info(" Aucune modification effectuée") + + +async def delete_swagger_user(username: str): + """Supprimer un utilisateur Swagger""" + async with async_session_factory() as session: + result = await session.execute( + select(SwaggerUser).where(SwaggerUser.username == username) + ) + user = result.scalar_one_or_none() + + if not user: + logger.error(f" Utilisateur '{username}' introuvable") + return + + await session.delete(user) + await session.commit() + logger.info(f"🗑️ Utilisateur Swagger supprimé: {username}") async def list_available_tags(): """Liste tous les tags disponibles avec description""" - logger.info("\n TAGS DISPONIBLES:\n") + logger.info("\n🏷️ TAGS DISPONIBLES:\n") logger.info("=" * 80) for tag, desc in AVAILABLE_TAGS.items(): @@ -302,20 +337,135 @@ async def list_available_tags(): logger.info("=" * 80) -async def delete_swagger_user(username: str): +async def create_api_key( + name: str, + description: str = None, + expires_in_days: int = 365, + rate_limit: int = 60, + endpoints: list = None, +): + """Créer une clé API""" async with async_session_factory() as session: - result = await session.execute( - select(SwaggerUser).where(SwaggerUser.username == username) - ) - user = result.scalar_one_or_none() + service = ApiKeyService(session) - if not user: - logger.error(f" Utilisateur '{username}' introuvable") + api_key_obj, api_key_plain = await service.create_api_key( + name=name, + description=description, + created_by="cli", + expires_in_days=expires_in_days, + rate_limit_per_minute=rate_limit, + allowed_endpoints=endpoints, + ) + + logger.info("=" * 70) + logger.info("🔑 Clé API créée avec succès") + logger.info("=" * 70) + logger.info(f" ID: {api_key_obj.id}") + logger.info(f" Nom: {api_key_obj.name}") + logger.info(f" Clé: {api_key_plain}") + logger.info(f" Préfixe: {api_key_obj.key_prefix}") + logger.info(f" Rate limit: {api_key_obj.rate_limit_per_minute} req/min") + logger.info(f" Expire le: {api_key_obj.expires_at}") + + if api_key_obj.allowed_endpoints: + try: + endpoints_list = json.loads(api_key_obj.allowed_endpoints) + logger.info(f" Endpoints: {', '.join(endpoints_list)}") + except Exception: + logger.info(f" Endpoints: {api_key_obj.allowed_endpoints}") + else: + logger.info(" Endpoints: Tous (aucune restriction)") + + logger.info("=" * 70) + logger.info(" SAUVEGARDEZ CETTE CLÉ - Elle ne sera plus affichée !") + logger.info("=" * 70) + + +async def list_api_keys(): + """Lister toutes les clés API""" + async with async_session_factory() as session: + service = ApiKeyService(session) + keys = await service.list_api_keys() + + if not keys: + logger.info("🔭 Aucune clé API") return - await session.delete(user) + logger.info(f"🔑 {len(keys)} clé(s) API:\n") + + for key in keys: + is_valid = key.is_active and ( + not key.expires_at or key.expires_at > datetime.now() + ) + status = "" if is_valid else "" + + logger.info(f" {status} {key.name:<30} ({key.key_prefix}...)") + logger.info(f" ID: {key.id}") + logger.info(f" Rate limit: {key.rate_limit_per_minute} req/min") + logger.info(f" Requêtes: {key.total_requests}") + logger.info(f" Expire: {key.expires_at or 'Jamais'}") + logger.info(f" Dernière utilisation: {key.last_used_at or 'Jamais'}") + + if key.allowed_endpoints: + try: + endpoints = json.loads(key.allowed_endpoints) + display = ", ".join(endpoints[:4]) + if len(endpoints) > 4: + display += f"... (+{len(endpoints) - 4})" + logger.info(f" Endpoints: {display}") + except Exception: + pass + else: + logger.info(" Endpoints: Tous") + logger.info("") + + +async def revoke_api_key(key_id: str): + """Révoquer une clé API""" + async with async_session_factory() as session: + result = await session.execute(select(ApiKey).where(ApiKey.id == key_id)) + key = result.scalar_one_or_none() + + if not key: + logger.error(f" Clé API '{key_id}' introuvable") + return + + key.is_active = False + key.revoked_at = datetime.now() await session.commit() - logger.info(f"🗑️ Utilisateur Swagger supprimé: {username}") + + logger.info(f"🗑️ Clé API révoquée: {key.name}") + logger.info(f" ID: {key.id}") + + +async def verify_api_key(api_key: str): + """Vérifier une clé API""" + async with async_session_factory() as session: + service = ApiKeyService(session) + key = await service.verify_api_key(api_key) + + if not key: + logger.error(" Clé API invalide ou expirée") + return + + logger.info("=" * 60) + logger.info(" Clé API valide") + logger.info("=" * 60) + logger.info(f" Nom: {key.name}") + logger.info(f" ID: {key.id}") + logger.info(f" Rate limit: {key.rate_limit_per_minute} req/min") + logger.info(f" Requêtes totales: {key.total_requests}") + logger.info(f" Expire: {key.expires_at or 'Jamais'}") + + if key.allowed_endpoints: + try: + endpoints = json.loads(key.allowed_endpoints) + logger.info(f" Endpoints autorisés: {endpoints}") + except Exception: + pass + else: + logger.info(" Endpoints autorisés: Tous") + logger.info("=" * 60) async def main(): @@ -325,6 +475,8 @@ async def main(): epilog=""" EXEMPLES D'UTILISATION: +=== UTILISATEURS SWAGGER === + 1. Créer un utilisateur avec preset: python scripts/manage_security.py swagger add commercial Pass123! --preset commercial @@ -348,6 +500,23 @@ python scripts/manage_security.py swagger tags 8. Désactiver temporairement: python scripts/manage_security.py swagger update client --inactive + +=== CLÉS API === + +9. Créer une clé API: +python scripts/manage_security.py apikey create "Mon App" --days 365 --rate-limit 100 + +10. Créer avec endpoints restreints: + python scripts/manage_security.py apikey create "SDK-ReadOnly" --endpoints "/clients" "/clients/*" "/devis" "/devis/*" + +11. Lister les clés: + python scripts/manage_security.py apikey list + +12. Vérifier une clé: + python scripts/manage_security.py apikey verify sdk_live_xxxxx + +13. Révoquer une clé: + python scripts/manage_security.py apikey revoke """, ) @@ -392,6 +561,24 @@ python scripts/manage_security.py swagger update client --inactive swagger_sub.add_parser("tags", help="Lister les tags disponibles") + apikey_parser = subparsers.add_parser("apikey", help="Gestion clés API") + apikey_sub = apikey_parser.add_subparsers(dest="apikey_command") + + create_p = apikey_sub.add_parser("create", help="Créer clé API") + create_p.add_argument("name", help="Nom de la clé") + create_p.add_argument("--description", help="Description") + create_p.add_argument("--days", type=int, default=365, help="Expiration (jours)") + create_p.add_argument("--rate-limit", type=int, default=60, help="Req/min") + create_p.add_argument("--endpoints", nargs="+", help="Endpoints autorisés") + + apikey_sub.add_parser("list", help="Lister clés") + + rev_p = apikey_sub.add_parser("revoke", help="Révoquer clé") + rev_p.add_argument("key_id", help="ID de la clé") + + ver_p = apikey_sub.add_parser("verify", help="Vérifier clé") + ver_p.add_argument("api_key", help="Clé API complète") + args = parser.parse_args() if not args.command: @@ -431,12 +618,30 @@ python scripts/manage_security.py swagger update client --inactive else: swagger_parser.print_help() + elif args.command == "apikey": + if args.apikey_command == "create": + await create_api_key( + name=args.name, + description=args.description, + expires_in_days=args.days, + rate_limit=args.rate_limit, + endpoints=args.endpoints, + ) + elif args.apikey_command == "list": + await list_api_keys() + elif args.apikey_command == "revoke": + await revoke_api_key(args.key_id) + elif args.apikey_command == "verify": + await verify_api_key(args.api_key) + else: + apikey_parser.print_help() + if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: - print("\nℹ️ Interrupted") + print("\n Interrupted") sys.exit(0) except Exception as e: logger.error(f" Erreur: {e}")