fanilo/Sage100-vps
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(scripts): improve logging format and endpoint handling in security management

Browse source
This commit is contained in:
Fanilo-Nantenaina 2026-01-20 15:01:57 +03:00
parent 3cdb490ee5
commit 1a08894b47
2 changed files with 50 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
middleware/security.py
View file

@ -4,7 +4,7 @@ from fastapi.security import HTTPBasic, HTTPBasicCredentials
from starlette.middleware.base import BaseHTTPMiddleware from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp from starlette.types import ASGIApp
from sqlalchemy import select from sqlalchemy import select
from typing import Callable from typing import Optional, Callable
from datetime import datetime from datetime import datetime
import logging import logging
import base64 import base64

80
scripts/manage_security.py
View file

@ -1,7 +1,6 @@
import asyncio import asyncio
import sys import sys
from pathlib import Path from pathlib import Path
from database import get_session from database import get_session
from database.models.api_key import SwaggerUser, ApiKey from database.models.api_key import SwaggerUser, ApiKey
from services.api_key import ApiKeyService from services.api_key import ApiKeyService
@ -31,7 +30,7 @@ async def add_swagger_user(username: str, password: str, full_name: str = None):
existing = result.scalar_one_or_none() existing = result.scalar_one_or_none()
if existing: if existing:
logger.error(f" L'utilisateur '{username}' existe déjà") logger.error(f"❌ L'utilisateur '{username}' existe déjà")
return return
swagger_user = SwaggerUser( swagger_user = SwaggerUser(
@ -44,7 +43,7 @@ async def add_swagger_user(username: str, password: str, full_name: str = None):
session.add(swagger_user) session.add(swagger_user)
await session.commit() await session.commit()
logger.info(f" Utilisateur Swagger créé: {username}") logger.info(f"✅ Utilisateur Swagger créé: {username}")
logger.info(f" Nom complet: {swagger_user.full_name}") logger.info(f" Nom complet: {swagger_user.full_name}")
logger.info(f" Actif: {swagger_user.is_active}") logger.info(f" Actif: {swagger_user.is_active}")
@ -59,13 +58,13 @@ async def list_swagger_users():
users = result.scalars().all() users = result.scalars().all()
if not users: if not users:
logger.info(" Aucun utilisateur Swagger") logger.info("📭 Aucun utilisateur Swagger")
break break
logger.info(f" {len(users)} utilisateur(s) Swagger:\n") logger.info(f"👥 {len(users)} utilisateur(s) Swagger:\n")
for user in users: for user in users:
status = "" if user.is_active else "" status = "✅" if user.is_active else "❌"
logger.info(f" {status} {user.username}") logger.info(f" {status} {user.username}")
logger.info(f" Nom: {user.full_name}") logger.info(f" Nom: {user.full_name}")
logger.info(f" Créé: {user.created_at}") logger.info(f" Créé: {user.created_at}")
@ -85,13 +84,13 @@ async def delete_swagger_user(username: str):
user = result.scalar_one_or_none() user = result.scalar_one_or_none()
if not user: if not user:
logger.error(f" Utilisateur '{username}' introuvable") logger.error(f"❌ Utilisateur '{username}' introuvable")
break break
await session.delete(user) await session.delete(user)
await session.commit() await session.commit()
logger.info(f" Utilisateur Swagger supprimé: {username}") logger.info(f"🗑️ Utilisateur Swagger supprimé: {username}")
break break
@ -116,9 +115,9 @@ async def create_api_key(
allowed_endpoints=endpoints, allowed_endpoints=endpoints,
) )
logger.info("=" * 60) logger.info("=" * 70)
logger.info(" Clé API créée avec succès") logger.info("🔑 Clé API créée avec succès")
logger.info("=" * 60) logger.info("=" * 70)
logger.info(f" ID: {api_key_obj.id}") logger.info(f" ID: {api_key_obj.id}")
logger.info(f" Nom: {api_key_obj.name}") logger.info(f" Nom: {api_key_obj.name}")
logger.info(f" Clé: {api_key_plain}") logger.info(f" Clé: {api_key_plain}")
@ -128,15 +127,16 @@ async def create_api_key(
logger.info(f" Expire le: {api_key_obj.expires_at}") logger.info(f" Expire le: {api_key_obj.expires_at}")
if api_key_obj.allowed_endpoints: if api_key_obj.allowed_endpoints:
logger.info( import json
f" Endpoints autorisés: {', '.join(api_key_obj.allowed_endpoints)}"
) endpoints_list = json.loads(api_key_obj.allowed_endpoints)
logger.info(f" Endpoints autorisés: {', '.join(endpoints_list)}")
else: else:
logger.info(" Endpoints autorisés: Tous") logger.info(" Endpoints autorisés: Tous")
logger.info("=" * 60) logger.info("=" * 70)
logger.info(" IMPORTANT: Sauvegardez cette clé, elle ne sera plus affichée !") logger.info("⚠️ IMPORTANT: Sauvegardez cette clé, elle ne sera plus affichée !")
logger.info("=" * 60) logger.info("=" * 70)
break break
@ -149,17 +149,17 @@ async def list_api_keys():
keys = await service.list_api_keys() keys = await service.list_api_keys()
if not keys: if not keys:
logger.info(" Aucune clé API") logger.info("📭 Aucune clé API")
break break
logger.info(f" {len(keys)} clé(s) API:\n") logger.info(f"🔑 {len(keys)} clé(s) API:\n")
for key in keys: for key in keys:
status = ( status = (
"" "✅"
if key.is_active if key.is_active
and (not key.expires_at or key.expires_at > datetime.now()) and (not key.expires_at or key.expires_at > datetime.now())
else "" else "❌"
) )
logger.info(f" {status} {key.name:<30} ({key.key_prefix}...)") logger.info(f" {status} {key.name:<30} ({key.key_prefix}...)")
@ -171,9 +171,15 @@ async def list_api_keys():
logger.info(f" Dernière utilisation: {key.last_used_at or 'Jamais'}") logger.info(f" Dernière utilisation: {key.last_used_at or 'Jamais'}")
if key.allowed_endpoints: if key.allowed_endpoints:
logger.info( import json
f" Endpoints: {', '.join(key.allowed_endpoints[:3])}..."
) try:
endpoints = json.loads(key.allowed_endpoints)
logger.info(
f" Endpoints: {', '.join(endpoints[:5])}{'...' if len(endpoints) > 5 else ''}"
)
except:
pass
logger.info("") logger.info("")
@ -190,13 +196,13 @@ async def revoke_api_key(key_id: str):
key = result.scalar_one_or_none() key = result.scalar_one_or_none()
if not key: if not key:
logger.error(f" Clé API '{key_id}' introuvable") logger.error(f"❌ Clé API '{key_id}' introuvable")
break break
key.is_active = False key.is_active = False
await session.commit() await session.commit()
logger.info(f" Clé API révoquée: {key.name}") logger.info(f"🗑️ Clé API révoquée: {key.name}")
logger.info(f" ID: {key.id}") logger.info(f" ID: {key.id}")
logger.info(f" Préfixe: {key.key_prefix}") logger.info(f" Préfixe: {key.key_prefix}")
@ -212,11 +218,11 @@ async def verify_api_key(api_key: str):
key = await service.verify_api_key(api_key) key = await service.verify_api_key(api_key)
if not key: if not key:
logger.error(" Clé API invalide ou expirée") logger.error("❌ Clé API invalide ou expirée")
break break
logger.info("=" * 60) logger.info("=" * 60)
logger.info(" Clé API valide") logger.info("✅ Clé API valide")
logger.info("=" * 60) logger.info("=" * 60)
logger.info(f" Nom: {key.name}") logger.info(f" Nom: {key.name}")
logger.info(f" ID: {key.id}") logger.info(f" ID: {key.id}")
@ -224,6 +230,18 @@ async def verify_api_key(api_key: str):
logger.info(f" Requêtes totales: {key.total_requests}") logger.info(f" Requêtes totales: {key.total_requests}")
logger.info(f" Expire le: {key.expires_at or 'Jamais'}") logger.info(f" Expire le: {key.expires_at or 'Jamais'}")
logger.info(f" Dernière utilisation: {key.last_used_at or 'Jamais'}") logger.info(f" Dernière utilisation: {key.last_used_at or 'Jamais'}")
if key.allowed_endpoints:
import json
try:
endpoints = json.loads(key.allowed_endpoints)
logger.info(f" Endpoints autorisés: {endpoints}")
except:
pass
else:
logger.info(" Endpoints autorisés: Tous")
logger.info("=" * 60) logger.info("=" * 60)
break break
@ -267,7 +285,7 @@ async def main():
create_parser.add_argument( create_parser.add_argument(
"--endpoints", "--endpoints",
nargs="+", nargs="+",
help="Endpoints autorisés (ex: /clients /articles)", help="Endpoints autorisés (ex: /clients /articles /devis/*)",
) )
apikey_subparsers.add_parser("list", help="Lister les clés API") apikey_subparsers.add_parser("list", help="Lister les clés API")
@ -317,10 +335,10 @@ if __name__ == "__main__":
try: try:
asyncio.run(main()) asyncio.run(main())
except KeyboardInterrupt: except KeyboardInterrupt:
logger.info("\n Interrupted") logger.info("\n⏹️ Interrupted")
sys.exit(0) sys.exit(0)
except Exception as e: except Exception as e:
logger.error(f" Erreur: {e}") logger.error(f"❌ Erreur: {e}")
import traceback import traceback
traceback.print_exc() traceback.print_exc()