""" Script para generar datos de prueba de seguridad en logs de auditoría. Esto permite probar la funcionalidad de análisis de seguridad con diferentes tipos de amenazas. """ import asyncio import sys from pathlib import Path from datetime import datetime, timedelta, timezone import uuid import random # Agregar el directorio raíz al path sys.path.insert(0, str(Path(__file__).parent.parent)) from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.core.database import AsyncSessionLocal from app.models.audit import AuditLog from app.models.user import User from app.models.tenant import Tenant async def generate_test_data(): """Genera logs de auditoría de prueba para análisis de seguridad.""" async with AsyncSessionLocal() as db: # Obtener tenant y usuarios de prueba tenant_result = await db.execute(select(Tenant).limit(1)) tenant = tenant_result.scalar_one_or_none() if not tenant: print("❌ No se encontró ningún tenant. Ejecuta las migraciones primero.") return user_result = await db.execute(select(User).where(User.tenant_id == tenant.id).limit(1)) user = user_result.scalar_one_or_none() if not user: print("❌ No se encontró ningún usuario. Crea un usuario primero.") return print(f"✅ Usando tenant: {tenant.name}") print(f"✅ Usando usuario: {user.email}") print() now = datetime.now(timezone.utc) # IPs de prueba suspicious_ips = [ "192.168.1.100", "10.0.0.50", "172.16.0.10", "203.0.113.42", "198.51.100.88" ] logs_created = 0 # ============================================ # 1. GENERAR INTENTOS FALLIDOS DE LOGIN (Fuerza Bruta) # ============================================ print("🔐 Generando intentos fallidos de login...") # Generar 25 intentos fallidos (esto hará que sea HIGH severity) for i in range(25): time_offset = timedelta(hours=random.randint(0, 23), minutes=random.randint(0, 59)) log = AuditLog( id=uuid.uuid4(), tenant_id=tenant.id, user_id=user.id, action="user.login_failed", resource_type="auth", resource_id=None, ip_address=random.choice(suspicious_ips), user_agent="Mozilla/5.0 (Test Browser)", metadata={"reason": "invalid_credentials", "username": f"test_user_{i}"}, created_at=now - time_offset ) db.add(log) logs_created += 1 print(f" ✓ Creados {25} intentos fallidos de login (HIGH severity)") # ============================================ # 2. GENERAR ELIMINACIONES MASIVAS (CRITICAL) # ============================================ print("🗑️ Generando eliminaciones masivas...") resources = ["ticket", "comment", "attachment", "category", "user"] # Generar 55 eliminaciones (esto hará que sea CRITICAL severity) for i in range(55): time_offset = timedelta(hours=random.randint(0, 23), minutes=random.randint(0, 59)) resource = random.choice(resources) log = AuditLog( id=uuid.uuid4(), tenant_id=tenant.id, user_id=user.id, action=f"{resource}.delete", resource_type=resource, resource_id=uuid.uuid4(), ip_address=random.choice(suspicious_ips), user_agent="Mozilla/5.0 (Test Browser)", metadata={"deleted_by": user.email}, created_at=now - time_offset ) db.add(log) logs_created += 1 print(f" ✓ Creadas {55} eliminaciones masivas (CRITICAL severity)") # ============================================ # 3. GENERAR CAMBIOS DE PRIVILEGIOS (HIGH) # ============================================ print("👤 Generando cambios de privilegios...") roles = ["AGENT", "CLIENT_USER", "AUDITOR", "SUPPORT_MANAGER", "ADMIN"] # Generar 5 cambios de rol (esto hará que sea HIGH severity) for i in range(5): time_offset = timedelta(hours=random.randint(0, 23), minutes=random.randint(0, 59)) old_role = random.choice(roles) new_role = random.choice([r for r in roles if r != old_role]) log = AuditLog( id=uuid.uuid4(), tenant_id=tenant.id, user_id=user.id, action="user.update", resource_type="user", resource_id=uuid.uuid4(), ip_address=random.choice(suspicious_ips), user_agent="Mozilla/5.0 (Test Browser)", old_values={"role": old_role}, new_values={"role": new_role}, metadata={"changed_by": user.email}, created_at=now - time_offset ) db.add(log) logs_created += 1 print(f" ✓ Creados {5} cambios de privilegios (HIGH severity)") # ============================================ # 4. GENERAR LOGS NORMALES (para dar contexto) # ============================================ print("📋 Generando logs de actividad normal...") normal_actions = [ "ticket.create", "ticket.update", "comment.create", "user.login", "ticket.view", ] for i in range(20): time_offset = timedelta(hours=random.randint(0, 23), minutes=random.randint(0, 59)) action = random.choice(normal_actions) log = AuditLog( id=uuid.uuid4(), tenant_id=tenant.id, user_id=user.id, action=action, resource_type=action.split('.')[0], resource_id=uuid.uuid4(), ip_address=random.choice(suspicious_ips), user_agent="Mozilla/5.0 (Test Browser)", metadata={"action": "normal_activity"}, created_at=now - time_offset ) db.add(log) logs_created += 1 print(f" ✓ Creados {20} logs de actividad normal") # Guardar todo await db.commit() print() print("=" * 60) print(f"✅ GENERACIÓN COMPLETADA") print(f" Total de logs creados: {logs_created}") print() print("📊 Amenazas esperadas en el análisis:") print(" 🔴 1 amenaza CRÍTICA: 55 eliminaciones masivas") print(" 🟠 1 amenaza HIGH: 25 intentos fallidos de login") print(" 🟠 1 amenaza HIGH: 5 cambios de privilegios") print() print("🌐 Accede a la página de seguridad para ver el análisis") print("=" * 60) async def cleanup_test_data(): """Elimina los logs de auditoría de prueba.""" async with AsyncSessionLocal() as db: tenant_result = await db.execute(select(Tenant).limit(1)) tenant = tenant_result.scalar_one_or_none() if not tenant: print("❌ No se encontró ningún tenant.") return # Eliminar logs de las últimas 24 horas now = datetime.now(timezone.utc) cutoff = now - timedelta(hours=24) result = await db.execute( select(AuditLog).where( AuditLog.tenant_id == tenant.id, AuditLog.created_at >= cutoff ) ) logs = result.scalars().all() if not logs: print("ℹ️ No hay logs de prueba para eliminar.") return for log in logs: await db.delete(log) await db.commit() print(f"✅ Eliminados {len(logs)} logs de prueba de las últimas 24 horas") if __name__ == "__main__": import sys if len(sys.argv) > 1 and sys.argv[1] == "cleanup": print("🧹 Limpiando datos de prueba...") asyncio.run(cleanup_test_data()) else: print("🚀 Generando datos de prueba para análisis de seguridad...") print() asyncio.run(generate_test_data()) print() print("💡 Para limpiar estos datos de prueba, ejecuta:") print(" python scripts/generate_security_test_data.py cleanup")