Compare commits
No commits in common. "437ecd0ed3f6053f7a301ac9366ecc9fa3d7d67e" and "23a94f5558307ecd5670b3f7747534574cf906d1" have entirely different histories.
437ecd0ed3
...
23a94f5558
4 changed files with 256 additions and 496 deletions
216
api.py
216
api.py
|
|
@ -170,160 +170,8 @@ def get_swagger_user_from_state(request: Request) -> Optional[dict]:
|
|||
return getattr(request.state, "swagger_user", None)
|
||||
|
||||
|
||||
def get_schemas_for_tags(allowed_tags: List[str], all_schemas: dict) -> dict:
|
||||
if not allowed_tags: # Admin = tous les schémas
|
||||
return all_schemas
|
||||
|
||||
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": {
|
||||
"type": "http",
|
||||
"scheme": "bearer",
|
||||
"bearerFormat": "JWT",
|
||||
"description": "Authentification JWT pour utilisateurs (POST /auth/login)",
|
||||
},
|
||||
"ApiKeyAuth": {
|
||||
"type": "apiKey",
|
||||
"in": "header",
|
||||
"name": "X-API-Key",
|
||||
"description": "Clé API pour intégrations externes (format: sdk_live_xxx)",
|
||||
},
|
||||
}
|
||||
|
||||
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
|
||||
app: FastAPI, allowed_tags: Optional[List[str]] = None
|
||||
) -> dict:
|
||||
base_schema = get_openapi(
|
||||
title=app.title,
|
||||
|
|
@ -332,36 +180,25 @@ def generate_filtered_openapi_schema(
|
|||
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"] = {
|
||||
"HTTPBearer": {
|
||||
"type": "http",
|
||||
"scheme": "bearer",
|
||||
"bearerFormat": "JWT",
|
||||
"description": "Authentification JWT pour utilisateurs (POST /auth/login)",
|
||||
},
|
||||
"ApiKeyAuth": {
|
||||
"type": "apiKey",
|
||||
"in": "header",
|
||||
"name": "X-API-Key",
|
||||
"description": "Clé API pour intégrations externes (format: sdk_live_xxx)",
|
||||
},
|
||||
}
|
||||
|
||||
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 []
|
||||
base_schema["security"] = [{"HTTPBearer": []}, {"ApiKeyAuth": []}]
|
||||
|
||||
if not allowed_tags:
|
||||
logger.info(" Schéma OpenAPI complet (admin)")
|
||||
logger.info("📚 Schéma OpenAPI complet (admin)")
|
||||
return base_schema
|
||||
|
||||
filtered_paths = {}
|
||||
|
|
@ -398,19 +235,7 @@ def generate_filtered_openapi_schema(
|
|||
if tag_obj.get("name") in 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}"
|
||||
)
|
||||
logger.info(f"🔒 Schéma filtré: {len(filtered_paths)} paths, tags: {allowed_tags}")
|
||||
|
||||
return base_schema
|
||||
|
||||
|
|
@ -429,9 +254,9 @@ async def custom_openapi_endpoint(request: Request):
|
|||
username = swagger_user.get("username", "unknown")
|
||||
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, swagger_user)
|
||||
schema = generate_filtered_openapi_schema(app, allowed_tags)
|
||||
|
||||
return JSONResponse(content=schema)
|
||||
|
||||
|
|
@ -456,7 +281,6 @@ async def custom_swagger_ui(request: Request):
|
|||
"displayRequestDuration": True,
|
||||
"filter": True,
|
||||
"tryItOutEnabled": True,
|
||||
"docExpansion": "list", # Meilleure UX
|
||||
},
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -152,7 +152,7 @@ templates_signature_email = {
|
|||
</table>
|
||||
|
||||
<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>,
|
||||
certifié eIDAS et conforme au RGPD. Votre identité sera vérifiée et le document sera
|
||||
horodaté de manière infalsifiable.
|
||||
|
|
|
|||
|
|
@ -64,7 +64,6 @@ class SwaggerAuthMiddleware:
|
|||
|
||||
if "state" not in scope:
|
||||
scope["state"] = {}
|
||||
|
||||
scope["state"]["swagger_user"] = swagger_user
|
||||
|
||||
logger.info(
|
||||
|
|
@ -72,7 +71,7 @@ class SwaggerAuthMiddleware:
|
|||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f" Erreur parsing auth header: {e}")
|
||||
logger.error(f"Erreur parsing auth header: {e}")
|
||||
response = JSONResponse(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
content={"detail": "Format d'authentification invalide"},
|
||||
|
|
@ -111,7 +110,7 @@ class SwaggerAuthMiddleware:
|
|||
return {
|
||||
"id": swagger_user.id,
|
||||
"username": swagger_user.username,
|
||||
"allowed_tags": swagger_user.allowed_tags_list,
|
||||
"allowed_tags": swagger_user.allowed_tags_list, # None = admin complet
|
||||
"is_active": swagger_user.is_active,
|
||||
}
|
||||
|
||||
|
|
@ -119,7 +118,7 @@ class SwaggerAuthMiddleware:
|
|||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f" Erreur vérification credentials: {e}", exc_info=True)
|
||||
logger.error(f"Erreur vérification credentials: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
|
|
@ -159,7 +158,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
|
|||
api_key_header = request.headers.get("X-API-Key")
|
||||
|
||||
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(
|
||||
request, api_key_header, path, method, call_next
|
||||
)
|
||||
|
|
@ -169,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
|
||||
|
|
@ -200,7 +199,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
|
|||
api_key_obj = await service.verify_api_key(api_key)
|
||||
|
||||
if not api_key_obj:
|
||||
logger.warning(f" Clé API invalide: {method} {path}")
|
||||
logger.warning(f"❌ Clé API invalide: {method} {path}")
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
content={
|
||||
|
|
@ -211,7 +210,7 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
|
|||
|
||||
is_allowed, rate_info = await service.check_rate_limit(api_key_obj)
|
||||
if not is_allowed:
|
||||
logger.warning(f"⏱️ Rate limit: {api_key_obj.name}")
|
||||
logger.warning(f"⏱ Rate limit: {api_key_obj.name}")
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
||||
content={"detail": "Rate limit dépassé"},
|
||||
|
|
@ -243,14 +242,14 @@ class ApiKeyMiddlewareHTTP(BaseHTTPMiddleware):
|
|||
"endpoint_requested": path,
|
||||
"api_key_name": api_key_obj.name,
|
||||
"allowed_endpoints": allowed,
|
||||
"hint": "Cette clé API n'a pas accès à cet endpoint.",
|
||||
"hint": "Cette clé API n'a pas accès à cet endpoint. Contactez l'administrateur.",
|
||||
},
|
||||
)
|
||||
|
||||
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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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,36 @@ _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,91 +51,20 @@ 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")
|
||||
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(
|
||||
username: str,
|
||||
password: str,
|
||||
full_name: str = None,
|
||||
tags: Optional[List[str]] = None,
|
||||
preset: Optional[str] = None,
|
||||
):
|
||||
"""Ajouter un utilisateur Swagger avec configuration avancée"""
|
||||
"""Ajouter un utilisateur Swagger"""
|
||||
async with async_session_factory() as session:
|
||||
result = await session.execute(
|
||||
select(SwaggerUser).where(SwaggerUser.username == username)
|
||||
|
|
@ -128,15 +75,6 @@ async def add_swagger_user(
|
|||
logger.error(f" L'utilisateur '{username}' existe déjà")
|
||||
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(
|
||||
username=username,
|
||||
hashed_password=hash_password(password),
|
||||
|
|
@ -151,17 +89,9 @@ async def add_swagger_user(
|
|||
logger.info(f" Utilisateur Swagger créé: {username}")
|
||||
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():
|
||||
"""Lister tous les utilisateurs Swagger avec détails"""
|
||||
"""Lister tous les utilisateurs Swagger"""
|
||||
async with async_session_factory() as session:
|
||||
result = await session.execute(select(SwaggerUser))
|
||||
users = result.scalars().all()
|
||||
|
|
@ -170,136 +100,27 @@ async def list_swagger_users():
|
|||
logger.info("🔭 Aucun utilisateur Swagger")
|
||||
return
|
||||
|
||||
logger.info(f"\n👥 {len(users)} utilisateur(s) Swagger:\n")
|
||||
logger.info("=" * 80)
|
||||
|
||||
logger.info("👥 {} utilisateur(s) Swagger:\n".format(len(users)))
|
||||
for user in users:
|
||||
status = " ACTIF" if user.is_active else " NON ACTIF"
|
||||
logger.info(f"\n{status} {user.username}")
|
||||
logger.info(f"📛 Nom: {user.full_name}")
|
||||
logger.info(f"🆔 ID: {user.id}")
|
||||
logger.info(f"📅 Créé: {user.created_at}")
|
||||
logger.info(f"🕐 Dernière connexion: {user.last_login or 'Jamais'}")
|
||||
status = "ACTIF" if user.is_active else "NON ACTIF"
|
||||
logger.info(" {} {}".format(status, user.username))
|
||||
logger.info("Nom: {}".format(user.full_name))
|
||||
logger.info("Créé: {}".format(user.created_at))
|
||||
logger.info(" Dernière connexion: {}".format(user.last_login or "Jamais"))
|
||||
|
||||
if user.allowed_tags:
|
||||
try:
|
||||
tags = json.loads(user.allowed_tags)
|
||||
if tags:
|
||||
logger.info(f"🏷️ Tags autorisés ({len(tags)}):")
|
||||
for tag in tags:
|
||||
desc = AVAILABLE_TAGS.get(tag, "")
|
||||
logger.info(f" • {tag} {desc}")
|
||||
|
||||
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)}"
|
||||
)
|
||||
logger.info("Tags autorisés: {}".format(", ".join(tags)))
|
||||
else:
|
||||
logger.info("👑 Tags autorisés: ADMIN COMPLET (tous)")
|
||||
logger.info("🔐 Authentification: JWT + X-API-Key (tout)")
|
||||
logger.info("Tags autorisés: Tous (admin)")
|
||||
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("Tags autorisés: Tous (admin)")
|
||||
|
||||
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)
|
||||
logger.info("")
|
||||
|
||||
|
||||
async def delete_swagger_user(username: str):
|
||||
|
|
@ -310,7 +131,7 @@ async def delete_swagger_user(username: str):
|
|||
user = result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
logger.error(f" Utilisateur '{username}' introuvable")
|
||||
logger.error(f"❌ Utilisateur '{username}' introuvable")
|
||||
return
|
||||
|
||||
await session.delete(user)
|
||||
|
|
@ -318,39 +139,161 @@ async def delete_swagger_user(username: str):
|
|||
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():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Gestion avancée des utilisateurs Swagger et clés API",
|
||||
description="Gestion des utilisateurs Swagger et clés API",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
EXEMPLES D'UTILISATION:
|
||||
|
||||
1. Créer un utilisateur avec preset:
|
||||
python scripts/manage_security.py swagger add commercial Pass123! --preset commercial
|
||||
|
||||
2. Créer un admin complet:
|
||||
python scripts/manage_security.py swagger add admin AdminPass
|
||||
|
||||
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
|
||||
Exemples:
|
||||
python scripts/manage_security.py swagger add admin MyP@ssw0rd
|
||||
python scripts/manage_security.py swagger list
|
||||
python scripts/manage_security.py apikey create "Mon App" --days 365 --rate-limit 100
|
||||
python scripts/manage_security.py apikey create "SDK-ReadOnly" --endpoints "/clients" "/clients/*" "/devis" "/devis/*"
|
||||
python scripts/manage_security.py apikey list
|
||||
python scripts/manage_security.py apikey verify sdk_live_xxxxx
|
||||
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
|
||||
""",
|
||||
)
|
||||
|
||||
subparsers = parser.add_subparsers(dest="command", help="Commandes")
|
||||
|
||||
swagger_parser = subparsers.add_parser("swagger", help="Gestion Swagger")
|
||||
|
|
@ -363,34 +306,32 @@ python scripts/manage_security.py swagger update client --inactive
|
|||
add_p.add_argument(
|
||||
"--tags",
|
||||
nargs="*",
|
||||
help="Tags autorisés. Vide = admin complet",
|
||||
help="Tags autorisés (Clients Devis etc). Vide ou omis = admin complet",
|
||||
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")
|
||||
|
||||
del_p = swagger_sub.add_parser("delete", help="Supprimer utilisateur")
|
||||
del_p.add_argument("username", help="Nom d'utilisateur")
|
||||
|
||||
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()
|
||||
|
||||
|
|
@ -400,37 +341,33 @@ python scripts/manage_security.py swagger update client --inactive
|
|||
|
||||
if args.command == "swagger":
|
||||
if args.swagger_command == "add":
|
||||
await add_swagger_user(
|
||||
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,
|
||||
)
|
||||
tags = args.tags if args.tags else None
|
||||
await add_swagger_user(args.username, args.password, args.full_name, tags)
|
||||
elif args.swagger_command == "list":
|
||||
await list_swagger_users()
|
||||
elif args.swagger_command == "delete":
|
||||
await delete_swagger_user(args.username)
|
||||
elif args.swagger_command == "tags":
|
||||
await list_available_tags()
|
||||
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:
|
||||
|
|
|
|||
Loading…
Reference in a new issue