Compare commits

...

2 commits

Author SHA1 Message Date
Fanilo-Nantenaina
437ecd0ed3 Merge branch 'feat/controlled_swagger_access' into main_2 2026-01-21 13:26:49 +03:00
Fanilo-Nantenaina
c057a085ed refactor(security): improve swagger user management and logging 2026-01-21 13:21:16 +03:00
4 changed files with 499 additions and 259 deletions

206
api.py
View file

@ -170,17 +170,115 @@ def get_swagger_user_from_state(request: Request) -> Optional[dict]:
return getattr(request.state, "swagger_user", None) return getattr(request.state, "swagger_user", None)
def generate_filtered_openapi_schema( def get_schemas_for_tags(allowed_tags: List[str], all_schemas: dict) -> dict:
app: FastAPI, allowed_tags: Optional[List[str]] = None if not allowed_tags: # Admin = tous les schémas
) -> dict: return all_schemas
base_schema = get_openapi(
title=app.title,
version=app.version,
description=app.description,
routes=app.routes,
)
base_schema["components"]["securitySchemes"] = { base_schemas = {
"HTTPValidationError",
"ValidationError",
"HTTPException",
"TokenResponse",
"Login",
"RegisterRequest",
}
tag_schemas_map = {
"Clients": {
"ClientDetails",
"ClientCreate",
"ClientUpdate",
"TiersDetails",
"Contact",
"ContactCreate",
"ContactUpdate",
},
"Articles": {
"Article",
"ArticleCreate",
"ArticleUpdate",
"Familles",
"FamilleCreate",
},
"Devis": {
"Devis",
"DevisRequest",
"DevisUpdate",
"LigneDocument",
"EmailEnvoi",
"RelanceDevis",
},
"Commandes": {"CommandeCreate", "CommandeUpdate", "LigneDocument"},
"Factures": {
"FactureCreate",
"FactureUpdate",
"ReglementFactureCreate",
"ReglementMultipleCreate",
},
"Livraisons": {"LivraisonCreate", "LivraisonUpdate"},
"Avoirs": {"AvoirCreate", "AvoirUpdate"},
"Fournisseurs": {
"FournisseurDetails",
"FournisseurCreate",
"FournisseurUpdate",
},
"Collaborateurs": {
"CollaborateurDetails",
"CollaborateurCreate",
"CollaborateurUpdate",
},
"Stock": {"EntreeStock", "SortieStock", "MouvementStock"},
"Emails": {"EmailEnvoi", "EmailBatch", "TemplateEmail", "TemplatePreview"},
"API Keys Management": {
"ApiKeyCreate",
"ApiKeyResponse",
"ApiKeyCreatedResponse",
"ApiKeyList",
},
}
allowed_schemas = base_schemas.copy()
for tag in allowed_tags:
if tag in tag_schemas_map:
allowed_schemas.update(tag_schemas_map[tag])
filtered_schemas = {}
for schema_name, schema_def in all_schemas.items():
if schema_name in allowed_schemas:
filtered_schemas[schema_name] = schema_def
if "$ref" in str(schema_def):
_add_referenced_schemas(schema_def, all_schemas, filtered_schemas)
return filtered_schemas
def _add_referenced_schemas(
schema_def: dict, all_schemas: dict, filtered_schemas: dict
):
"""Ajoute récursivement les schémas référencés"""
if isinstance(schema_def, dict):
for key, value in schema_def.items():
if key == "$ref" and isinstance(value, str):
ref_name = value.split("/")[-1]
if ref_name in all_schemas and ref_name not in filtered_schemas:
filtered_schemas[ref_name] = all_schemas[ref_name]
_add_referenced_schemas(
all_schemas[ref_name], all_schemas, filtered_schemas
)
elif isinstance(value, (dict, list)):
_add_referenced_schemas(value, all_schemas, filtered_schemas)
elif isinstance(schema_def, list):
for item in schema_def:
_add_referenced_schemas(item, all_schemas, filtered_schemas)
def get_auth_schemes_for_user(swagger_user: dict) -> dict:
allowed_tags = swagger_user.get("allowed_tags")
if not allowed_tags:
return {
"HTTPBearer": { "HTTPBearer": {
"type": "http", "type": "http",
"scheme": "bearer", "scheme": "bearer",
@ -195,10 +293,75 @@ def generate_filtered_openapi_schema(
}, },
} }
base_schema["security"] = [{"HTTPBearer": []}, {"ApiKeyAuth": []}] schemes = {}
if "Authentication" in allowed_tags:
schemes["HTTPBearer"] = {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT",
"description": "Authentification JWT pour utilisateurs (POST /auth/login)",
}
if "API Keys Management" in allowed_tags or len(allowed_tags) > 3:
schemes["ApiKeyAuth"] = {
"type": "apiKey",
"in": "header",
"name": "X-API-Key",
"description": "Clé API pour intégrations externes (format: sdk_live_xxx)",
}
if not schemes:
schemes["HTTPBearer"] = {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT",
"description": "Authentification requise",
}
return schemes
def generate_filtered_openapi_schema(
app: FastAPI, allowed_tags: Optional[List[str]] = None, swagger_user: dict = None
) -> dict:
base_schema = get_openapi(
title=app.title,
version=app.version,
description=app.description,
routes=app.routes,
)
if swagger_user:
auth_schemes = get_auth_schemes_for_user(swagger_user)
else:
auth_schemes = {
"HTTPBearer": {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT",
"description": "Authentification JWT",
},
"ApiKeyAuth": {
"type": "apiKey",
"in": "header",
"name": "X-API-Key",
"description": "Clé API",
},
}
base_schema["components"]["securitySchemes"] = auth_schemes
security_requirements = []
if "HTTPBearer" in auth_schemes:
security_requirements.append({"HTTPBearer": []})
if "ApiKeyAuth" in auth_schemes:
security_requirements.append({"ApiKeyAuth": []})
base_schema["security"] = security_requirements if security_requirements else []
if not allowed_tags: if not allowed_tags:
logger.info("📚 Schéma OpenAPI complet (admin)") logger.info(" Schéma OpenAPI complet (admin)")
return base_schema return base_schema
filtered_paths = {} filtered_paths = {}
@ -235,7 +398,19 @@ def generate_filtered_openapi_schema(
if tag_obj.get("name") in allowed_tags if tag_obj.get("name") in allowed_tags
] ]
logger.info(f"🔒 Schéma filtré: {len(filtered_paths)} paths, tags: {allowed_tags}") if "components" in base_schema and "schemas" in base_schema["components"]:
all_schemas = base_schema["components"]["schemas"]
filtered_schemas = get_schemas_for_tags(allowed_tags, all_schemas)
base_schema["components"]["schemas"] = filtered_schemas
logger.info(
f" Schéma filtré: {len(filtered_paths)} paths, "
f"{len(filtered_schemas)} schémas, tags: {allowed_tags}"
)
else:
logger.info(
f" Schéma filtré: {len(filtered_paths)} paths, tags: {allowed_tags}"
)
return base_schema return base_schema
@ -254,9 +429,9 @@ async def custom_openapi_endpoint(request: Request):
username = swagger_user.get("username", "unknown") username = swagger_user.get("username", "unknown")
allowed_tags = swagger_user.get("allowed_tags") allowed_tags = swagger_user.get("allowed_tags")
logger.info(f"📖 OpenAPI demandé par: {username}, tags: {allowed_tags or 'ALL'}") logger.info(f" OpenAPI demandé par: {username}, tags: {allowed_tags or 'ALL'}")
schema = generate_filtered_openapi_schema(app, allowed_tags) schema = generate_filtered_openapi_schema(app, allowed_tags, swagger_user)
return JSONResponse(content=schema) return JSONResponse(content=schema)
@ -281,6 +456,7 @@ async def custom_swagger_ui(request: Request):
"displayRequestDuration": True, "displayRequestDuration": True,
"filter": True, "filter": True,
"tryItOutEnabled": True, "tryItOutEnabled": True,
"docExpansion": "list", # Meilleure UX
}, },
) )

View file

@ -152,7 +152,7 @@ templates_signature_email = {
</table> </table>
<p style="color: #718096; font-size: 13px; line-height: 1.5; margin: 0;"> <p style="color: #718096; font-size: 13px; line-height: 1.5; margin: 0;">
<strong>🔒 Signature électronique sécurisée</strong><br> <strong> Signature électronique sécurisée</strong><br>
Votre signature est protégée par notre partenaire de confiance <strong>Universign</strong>, Votre signature est protégée par notre partenaire de confiance <strong>Universign</strong>,
certifié eIDAS et conforme au RGPD. Votre identité sera vérifiée et le document sera certifié eIDAS et conforme au RGPD. Votre identité sera vérifiée et le document sera
horodaté de manière infalsifiable. horodaté de manière infalsifiable.

View file

@ -64,6 +64,7 @@ class SwaggerAuthMiddleware:
if "state" not in scope: if "state" not in scope:
scope["state"] = {} scope["state"] = {}
scope["state"]["swagger_user"] = swagger_user scope["state"]["swagger_user"] = swagger_user
logger.info( logger.info(
@ -110,7 +111,7 @@ class SwaggerAuthMiddleware:
return { return {
"id": swagger_user.id, "id": swagger_user.id,
"username": swagger_user.username, "username": swagger_user.username,
"allowed_tags": swagger_user.allowed_tags_list, # None = admin complet "allowed_tags": swagger_user.allowed_tags_list,
"is_active": swagger_user.is_active, "is_active": swagger_user.is_active,
} }
@ -158,7 +159,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
api_key_header = request.headers.get("X-API-Key") api_key_header = request.headers.get("X-API-Key")
if api_key_header: if api_key_header:
logger.debug(f"🔑 API Key détectée pour {method} {path}") logger.debug(f" API Key détectée pour {method} {path}")
return await self._handle_api_key_auth( return await self._handle_api_key_auth(
request, api_key_header, path, method, call_next request, api_key_header, path, method, call_next
) )
@ -168,7 +169,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
if token.startswith("sdk_live_"): if token.startswith("sdk_live_"):
logger.warning( 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( return await self._handle_api_key_auth(
request, token, path, method, call_next request, token, path, method, call_next
@ -199,7 +200,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
api_key_obj = await service.verify_api_key(api_key) api_key_obj = await service.verify_api_key(api_key)
if not api_key_obj: if not api_key_obj:
logger.warning(f" Clé API invalide: {method} {path}") logger.warning(f" Clé API invalide: {method} {path}")
return JSONResponse( return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
content={ content={
@ -210,7 +211,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
is_allowed, rate_info = await service.check_rate_limit(api_key_obj) is_allowed, rate_info = await service.check_rate_limit(api_key_obj)
if not is_allowed: if not is_allowed:
logger.warning(f" Rate limit: {api_key_obj.name}") logger.warning(f" Rate limit: {api_key_obj.name}")
return JSONResponse( return JSONResponse(
status_code=status.HTTP_429_TOO_MANY_REQUESTS, status_code=status.HTTP_429_TOO_MANY_REQUESTS,
content={"detail": "Rate limit dépassé"}, content={"detail": "Rate limit dépassé"},
@ -242,14 +243,14 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
"endpoint_requested": path, "endpoint_requested": path,
"api_key_name": api_key_obj.name, "api_key_name": api_key_obj.name,
"allowed_endpoints": allowed, "allowed_endpoints": allowed,
"hint": "Cette clé API n'a pas accès à cet endpoint. Contactez l'administrateur.", "hint": "Cette clé API n'a pas accès à cet endpoint.",
}, },
) )
request.state.api_key = api_key_obj request.state.api_key = api_key_obj
request.state.authenticated_via = "api_key" 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) return await call_next(request)

View file

@ -1,3 +1,9 @@
#!/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 sys
import os import os
from pathlib import Path from pathlib import Path
@ -13,36 +19,12 @@ _current_file = Path(__file__).resolve()
_script_dir = _current_file.parent _script_dir = _current_file.parent
_app_dir = _script_dir.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: if str(_app_dir) in sys.path:
sys.path.remove(str(_app_dir)) sys.path.remove(str(_app_dir))
sys.path.insert(0, str(_app_dir)) sys.path.insert(0, str(_app_dir))
os.chdir(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: try:
from database.db_config import async_session_factory from database.db_config import async_session_factory
from database.models.api_key import SwaggerUser, ApiKey from database.models.api_key import SwaggerUser, ApiKey
@ -51,20 +33,91 @@ try:
except ImportError as e: except ImportError as e:
print(f"\n ERREUR D'IMPORT: {e}") print(f"\n ERREUR D'IMPORT: {e}")
print(" Vérifiez que vous êtes dans /app") print(" Vérifiez que vous êtes dans /app")
print(" Commande correcte: cd /app && python scripts/manage_security.py ...")
sys.exit(1) sys.exit(1)
logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s") logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s")
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
AVAILABLE_TAGS = {
"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",
"Tiers": "📋 Gestion générale des tiers",
"Contacts": "📞 Contacts des tiers",
"Articles": "📦 Catalogue articles",
"Familles": "🏷️ Familles d'articles",
"Stock": "📊 Mouvements de stock",
"Devis": "📄 Devis",
"Commandes": "🛒 Commandes",
"Livraisons": "🚚 Bons de livraison",
"Factures": "💰 Factures",
"Avoirs": "↩️ Avoirs",
"Règlements": "💳 Règlements et encaissements",
"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",
"System": "⚙️ Système et santé",
"Admin": "🛠️ Administration",
"Debug": "🐛 Debug et diagnostics",
}
PRESET_PROFILES = {
"commercial": [
"Clients",
"Contacts",
"Devis",
"Commandes",
"Factures",
"Articles",
"Documents",
"Emails",
],
"comptable": [
"Clients",
"Fournisseurs",
"Factures",
"Avoirs",
"Règlements",
"Documents",
"Emails",
],
"logistique": [
"Articles",
"Stock",
"Commandes",
"Livraisons",
"Fournisseurs",
"Documents",
],
"readonly": ["Clients", "Articles", "Devis", "Commandes", "Factures", "Documents"],
"developer": [
"Authentication",
"API Keys Management",
"System",
"Clients",
"Articles",
"Devis",
"Commandes",
"Factures",
],
}
async def add_swagger_user( async def add_swagger_user(
username: str, username: str,
password: str, password: str,
full_name: str = None, full_name: str = None,
tags: Optional[List[str]] = None, tags: Optional[List[str]] = None,
preset: Optional[str] = None,
): ):
"""Ajouter un utilisateur Swagger""" """Ajouter un utilisateur Swagger avec configuration avancée"""
async with async_session_factory() as session: async with async_session_factory() as session:
result = await session.execute( result = await session.execute(
select(SwaggerUser).where(SwaggerUser.username == username) select(SwaggerUser).where(SwaggerUser.username == username)
@ -75,6 +128,15 @@ async def add_swagger_user(
logger.error(f" L'utilisateur '{username}' existe déjà") logger.error(f" L'utilisateur '{username}' existe déjà")
return return
if preset:
if preset not in PRESET_PROFILES:
logger.error(
f" Preset '{preset}' inconnu. Disponibles: {list(PRESET_PROFILES.keys())}"
)
return
tags = PRESET_PROFILES[preset]
logger.info(f"📋 Application du preset '{preset}': {len(tags)} tags")
swagger_user = SwaggerUser( swagger_user = SwaggerUser(
username=username, username=username,
hashed_password=hash_password(password), hashed_password=hash_password(password),
@ -89,9 +151,17 @@ async def add_swagger_user(
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}")
if tags:
logger.info(f" 🏷️ Tags autorisés ({len(tags)}):")
for tag in tags:
desc = AVAILABLE_TAGS.get(tag, "")
logger.info(f"{tag} {desc}")
else:
logger.info(" 👑 Accès ADMIN COMPLET (tous les tags)")
async def list_swagger_users(): async def list_swagger_users():
"""Lister tous les utilisateurs Swagger""" """Lister tous les utilisateurs Swagger avec détails"""
async with async_session_factory() as session: async with async_session_factory() as session:
result = await session.execute(select(SwaggerUser)) result = await session.execute(select(SwaggerUser))
users = result.scalars().all() users = result.scalars().all()
@ -100,27 +170,136 @@ async def list_swagger_users():
logger.info("🔭 Aucun utilisateur Swagger") logger.info("🔭 Aucun utilisateur Swagger")
return return
logger.info("👥 {} utilisateur(s) Swagger:\n".format(len(users))) logger.info(f"\n👥 {len(users)} utilisateur(s) Swagger:\n")
logger.info("=" * 80)
for user in users: for user in users:
status = " ACTIF" if user.is_active else " NON ACTIF" status = " ACTIF" if user.is_active else " NON ACTIF"
logger.info(" {} {}".format(status, user.username)) logger.info(f"\n{status} {user.username}")
logger.info("Nom: {}".format(user.full_name)) logger.info(f"📛 Nom: {user.full_name}")
logger.info("Créé: {}".format(user.created_at)) logger.info(f"🆔 ID: {user.id}")
logger.info(" Dernière connexion: {}".format(user.last_login or "Jamais")) logger.info(f"📅 Créé: {user.created_at}")
logger.info(f"🕐 Dernière connexion: {user.last_login or 'Jamais'}")
if user.allowed_tags: if user.allowed_tags:
try: try:
tags = json.loads(user.allowed_tags) tags = json.loads(user.allowed_tags)
if tags: if tags:
logger.info("Tags autorisés: {}".format(", ".join(tags))) logger.info(f"🏷️ Tags autorisés ({len(tags)}):")
else: for tag in tags:
logger.info("Tags autorisés: Tous (admin)") desc = AVAILABLE_TAGS.get(tag, "")
except json.JSONDecodeError: logger.info(f"{tag} {desc}")
logger.info("Tags: Erreur format")
else:
logger.info("Tags autorisés: Tous (admin)")
logger.info("") auth_schemes = []
if "Authentication" in tags:
auth_schemes.append("JWT (Bearer)")
if "API Keys Management" in tags or len(tags) > 3:
auth_schemes.append("X-API-Key")
if not auth_schemes:
auth_schemes.append("JWT (Bearer)")
logger.info(
f"🔐 Authentification autorisée: {', '.join(auth_schemes)}"
)
else:
logger.info("👑 Tags autorisés: ADMIN COMPLET (tous)")
logger.info("🔐 Authentification: JWT + X-API-Key (tout)")
except json.JSONDecodeError:
logger.info("⚠️ Tags: Erreur format")
else:
logger.info("👑 Tags autorisés: ADMIN COMPLET (tous)")
logger.info("🔐 Authentification: JWT + X-API-Key (tout)")
logger.info("\n" + "=" * 80)
async def update_swagger_user(
username: str,
add_tags: Optional[List[str]] = None,
remove_tags: Optional[List[str]] = None,
set_tags: Optional[List[str]] = None,
preset: Optional[str] = None,
active: Optional[bool] = None,
):
"""Mettre à jour 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
modified = False
if preset:
if preset not in PRESET_PROFILES:
logger.error(f" Preset '{preset}' inconnu")
return
user.allowed_tags = json.dumps(PRESET_PROFILES[preset])
logger.info(f"📋 Preset '{preset}' appliqué")
modified = True
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}")
modified = True
elif add_tags or remove_tags:
current_tags = []
if user.allowed_tags:
try:
current_tags = json.loads(user.allowed_tags)
except json.JSONDecodeError:
current_tags = []
if add_tags:
for tag in add_tags:
if tag not in current_tags:
current_tags.append(tag)
logger.info(f" Tag ajouté: {tag}")
modified = True
if remove_tags:
for tag in remove_tags:
if tag in current_tags:
current_tags.remove(tag)
logger.info(f" Tag retiré: {tag}")
modified = True
user.allowed_tags = json.dumps(current_tags) if current_tags else None
if active is not None:
user.is_active = active
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")
async def list_available_tags():
"""Liste tous les tags disponibles avec description"""
logger.info("\n TAGS DISPONIBLES:\n")
logger.info("=" * 80)
for tag, desc in AVAILABLE_TAGS.items():
logger.info(f" {desc}")
logger.info(f" Nom: {tag}\n")
logger.info("=" * 80)
logger.info("\n📦 PRESETS DISPONIBLES:\n")
for preset_name, tags in PRESET_PROFILES.items():
logger.info(f" {preset_name}:")
logger.info(f" {', '.join(tags)}\n")
logger.info("=" * 80)
async def delete_swagger_user(username: str): async def delete_swagger_user(username: str):
@ -131,7 +310,7 @@ 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")
return return
await session.delete(user) await session.delete(user)
@ -139,161 +318,39 @@ async def delete_swagger_user(username: str):
logger.info(f"🗑️ Utilisateur Swagger supprimé: {username}") logger.info(f"🗑️ Utilisateur Swagger supprimé: {username}")
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:
service = ApiKeyService(session)
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(" ID: {}".format(api_key_obj.id))
logger.info(" Nom: {}".format(api_key_obj.name))
logger.info(" Clé: {}".format(api_key_plain))
logger.info(" Préfixe: {}".format(api_key_obj.key_prefix))
logger.info(
" Rate limit: {} req/min".format(api_key_obj.rate_limit_per_minute)
)
logger.info(" Expire le: {}".format(api_key_obj.expires_at))
if api_key_obj.allowed_endpoints:
import json
try:
endpoints_list = json.loads(api_key_obj.allowed_endpoints)
logger.info(" Endpoints: {}".format(", ".join(endpoints_list)))
except Exception:
logger.info(" Endpoints: {}".format(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
logger.info("🔑 {} clé(s) API:\n".format(len(keys)))
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:
import json
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"🗑️ 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:
import json
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(): async def main():
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="Gestion des utilisateurs Swagger et clés API", description="Gestion avancée des utilisateurs Swagger et clés API",
formatter_class=argparse.RawDescriptionHelpFormatter, formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=""" epilog="""
Exemples: EXEMPLES D'UTILISATION:
python scripts/manage_security.py swagger add admin MyP@ssw0rd
python scripts/manage_security.py swagger list 1. Créer un utilisateur avec preset:
python scripts/manage_security.py apikey create "Mon App" --days 365 --rate-limit 100 python scripts/manage_security.py swagger add commercial Pass123! --preset commercial
python scripts/manage_security.py apikey create "SDK-ReadOnly" --endpoints "/clients" "/clients/*" "/devis" "/devis/*"
python scripts/manage_security.py apikey list 2. Créer un admin complet:
python scripts/manage_security.py apikey verify sdk_live_xxxxx python scripts/manage_security.py swagger add admin AdminPass
python scripts/manage_security.py swagger add client_user Secret123 --full-name "Client Tech IT" --tags Authentication Clients Devis Factures
python scripts/manage_security.py swagger add admin_user AdminPass --tags # vide = tout voir 3. Créer avec tags spécifiques:
python scripts/manage_security.py swagger add client Pass123! --tags Clients Devis Factures
4. Mettre à jour un utilisateur (ajouter des tags):
python scripts/manage_security.py swagger update client --add-tags Commandes Livraisons
5. Changer complètement les tags:
python scripts/manage_security.py swagger update client --set-tags Clients Articles
6. Appliquer un preset:
python scripts/manage_security.py swagger update client --preset comptable
7. Lister les tags disponibles:
python scripts/manage_security.py swagger tags
8. Désactiver temporairement:
python scripts/manage_security.py swagger update client --inactive
""", """,
) )
subparsers = parser.add_subparsers(dest="command", help="Commandes") subparsers = parser.add_subparsers(dest="command", help="Commandes")
swagger_parser = subparsers.add_parser("swagger", help="Gestion Swagger") swagger_parser = subparsers.add_parser("swagger", help="Gestion Swagger")
@ -306,32 +363,34 @@ Exemples:
add_p.add_argument( add_p.add_argument(
"--tags", "--tags",
nargs="*", nargs="*",
help="Tags autorisés (Clients Devis etc). Vide ou omis = admin complet", help="Tags autorisés. Vide = admin complet",
default=None, default=None,
) )
add_p.add_argument(
"--preset",
choices=list(PRESET_PROFILES.keys()),
help="Appliquer un preset de tags",
)
update_p = swagger_sub.add_parser("update", help="Mettre à jour utilisateur")
update_p.add_argument("username", help="Nom d'utilisateur")
update_p.add_argument("--add-tags", nargs="+", help="Ajouter des tags")
update_p.add_argument("--remove-tags", nargs="+", help="Retirer des tags")
update_p.add_argument("--set-tags", nargs="*", help="Définir les tags (remplace)")
update_p.add_argument(
"--preset", choices=list(PRESET_PROFILES.keys()), help="Appliquer preset"
)
update_p.add_argument("--active", action="store_true", help="Activer l'utilisateur")
update_p.add_argument(
"--inactive", action="store_true", help="Désactiver l'utilisateur"
)
swagger_sub.add_parser("list", help="Lister utilisateurs") swagger_sub.add_parser("list", help="Lister utilisateurs")
del_p = swagger_sub.add_parser("delete", help="Supprimer utilisateur") del_p = swagger_sub.add_parser("delete", help="Supprimer utilisateur")
del_p.add_argument("username", help="Nom d'utilisateur") del_p.add_argument("username", help="Nom d'utilisateur")
apikey_parser = subparsers.add_parser("apikey", help="Gestion clés API") swagger_sub.add_parser("tags", help="Lister les tags disponibles")
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() args = parser.parse_args()
@ -341,33 +400,37 @@ Exemples:
if args.command == "swagger": if args.command == "swagger":
if args.swagger_command == "add": if args.swagger_command == "add":
tags = args.tags if args.tags else None await add_swagger_user(
await add_swagger_user(args.username, args.password, args.full_name, tags) args.username,
args.password,
args.full_name,
args.tags,
args.preset,
)
elif args.swagger_command == "update":
active = None
if args.active:
active = True
elif args.inactive:
active = False
await update_swagger_user(
args.username,
add_tags=args.add_tags,
remove_tags=args.remove_tags,
set_tags=args.set_tags,
preset=args.preset,
active=active,
)
elif args.swagger_command == "list": elif args.swagger_command == "list":
await list_swagger_users() await list_swagger_users()
elif args.swagger_command == "delete": elif args.swagger_command == "delete":
await delete_swagger_user(args.username) await delete_swagger_user(args.username)
elif args.swagger_command == "tags":
await list_available_tags()
else: else:
swagger_parser.print_help() 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__": if __name__ == "__main__":
try: try: