service_manager/backend/scripts/generate_security_test_data.py

"""
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")