Improve IP blocking and log analysis performance

Optimize critical IP blocking using bulk operations and refine log query for better accuracy and performance.

Replit-Commit-Author: Agent
Replit-Commit-Session-Id: 7a657272-55ba-4a79-9a2e-f1ed9bc7a528
Replit-Commit-Checkpoint-Type: intermediate_checkpoint
Replit-Commit-Event-Id: 545f1e17-399b-4078-b609-9458832db9c4
Replit-Commit-Screenshot-Url: https://storage.googleapis.com/screenshot-production-us-central1/449cf7c4-c97a-45ae-8234-e5c5b8d6a84f/7a657272-55ba-4a79-9a2e-f1ed9bc7a528/B8f0CIv

client/src/pages/Dashboard.tsx

@@ -206,7 +206,7 @@ export default function Dashboard() {
               {stats?.logs.recent || 0}
             </div>
             <p className="text-xs text-muted-foreground">
-              Ultimi 1000 log analizzati
+              Ultime 24 ore
             </p>
           </CardContent>
         </Card>

python_ml/main.py

@@ -505,7 +505,7 @@ async def block_ip(request: BlockIPRequest):
 
 @app.post("/block-all-critical")
 async def block_all_critical(request: BlockAllCriticalRequest):
-    """Blocca tutti gli IP critici non ancora bloccati sui router"""
+    """Blocca tutti gli IP critici non ancora bloccati sui router - versione ottimizzata con bulk blocking"""
     try:
         conn = get_db_connection()
         cursor = conn.cursor(cursor_factory=RealDictCursor)
@@ -514,6 +514,8 @@ async def block_all_critical(request: BlockAllCriticalRequest):
         routers = cursor.fetchall()
         
         if not routers:
+            cursor.close()
+            conn.close()
             raise HTTPException(status_code=400, detail="Nessun router configurato")
         
         cursor.execute("""
@@ -541,49 +543,63 @@ async def block_all_critical(request: BlockAllCriticalRequest):
                 "skipped_whitelisted": 0
             }
         
+        ip_data = {row['source_ip']: row for row in unblocked_ips}
+        ip_list = [row['source_ip'] for row in unblocked_ips]
+        
+        print(f"[BLOCK-ALL] Avvio blocco massivo: {len(ip_list)} IP con score >= {request.min_score} su {len(routers)} router")
+        
+        bulk_results = await mikrotik_manager.bulk_block_ips_on_all_routers(
+            routers=routers,
+            ip_list=ip_list,
+            list_name=request.list_name,
+            comment_prefix=f"IDS bulk-block (score>={request.min_score})",
+            timeout_duration="1h",
+            concurrency=10
+        )
+        
         blocked_count = 0
         failed_count = 0
         results_detail = []
         
-        for ip_row in unblocked_ips:
+        blocked_source_ips = []
-            ip = ip_row['source_ip']
+        
-            score = ip_row['max_score']
+        for ip in ip_list:
-            anomaly = ip_row['anomaly_type']
+            router_results = bulk_results.get(ip, {})
-            
+            score = ip_data[ip]['max_score']
-            try:
+            anomaly = ip_data[ip]['anomaly_type']
-                block_results = await mikrotik_manager.block_ip_on_all_routers(
+            
-                    routers,
+            if any(router_results.values()):
-                    ip,
+                blocked_count += 1
-                    list_name=request.list_name,
+                blocked_source_ips.append(ip)
-                    comment=f"IDS bulk-block: {anomaly} (score: {score:.0f})"
+                results_detail.append({"ip": ip, "score": float(score), "status": "blocked"})
-                )
+            else:
-                
+                failed_count += 1
-                if any(block_results.values()):
+                results_detail.append({"ip": ip, "score": float(score), "status": "failed"})
-                    cursor.execute("""
+        
-                        UPDATE detections 
+        if blocked_source_ips:
-                        SET blocked = true, blocked_at = NOW()
+            batch_size = 100
-                        WHERE source_ip = %s AND blocked = false
+            for i in range(0, len(blocked_source_ips), batch_size):
-                    """, (ip,))
+                batch = blocked_source_ips[i:i+batch_size]
-                    blocked_count += 1
+                placeholders = ','.join(['%s'] * len(batch))
-                    results_detail.append({"ip": ip, "score": score, "status": "blocked"})
+                cursor.execute(f"""
-                else:
+                    UPDATE detections 
-                    failed_count += 1
+                    SET blocked = true, blocked_at = NOW()
-                    results_detail.append({"ip": ip, "score": score, "status": "failed"})
+                    WHERE source_ip IN ({placeholders}) AND blocked = false
-                    
+                """, batch)
-            except Exception as e:
+            conn.commit()
-                failed_count += 1
+            print(f"[BLOCK-ALL] Database aggiornato: {len(blocked_source_ips)} IP marcati come bloccati")
-                results_detail.append({"ip": ip, "score": score, "status": f"error: {str(e)}"})
         
-        conn.commit()
         cursor.close()
         conn.close()
         
+        print(f"[BLOCK-ALL] Completato: {blocked_count} bloccati, {failed_count} falliti su {len(ip_list)} totali")
+        
         return {
             "message": f"Blocco massivo completato: {blocked_count} IP bloccati, {failed_count} falliti",
             "blocked": blocked_count,
             "failed": failed_count,
             "total_critical": len(unblocked_ips),
-            "details": results_detail[:50]
+            "details": results_detail[:100]
         }
         
     except HTTPException:

python_ml/mikrotik_manager.py

@@ -1,14 +1,14 @@
 """
 MikroTik Manager - Gestione router tramite API REST
 Più veloce e affidabile di SSH per 10+ router
+Porte REST API: 80 (HTTP) o 443 (HTTPS)
 """
 
 import httpx
 import asyncio
 import ssl
-from typing import List, Dict, Optional
+from typing import List, Dict, Optional, Set
 from datetime import datetime
-import hashlib
 import base64
 
 
@@ -16,39 +16,33 @@ class MikroTikManager:
     """
     Gestisce comunicazione con router MikroTik tramite API REST
     Supporta operazioni parallele su multipli router
+    Porte default: 80 (HTTP REST) o 443 (HTTPS REST)
     """
     
-    def __init__(self, timeout: int = 10):
+    def __init__(self, timeout: int = 15):
         self.timeout = timeout
-        self.clients = {}  # Cache di client HTTP per router
+        self.clients = {}
     
-    def _get_client(self, router_ip: str, username: str, password: str, port: int = 8728, use_ssl: bool = False) -> httpx.AsyncClient:
+    def _get_client(self, router_ip: str, username: str, password: str, port: int = 80, use_ssl: bool = False) -> httpx.AsyncClient:
         """Ottiene o crea client HTTP per un router"""
         key = f"{router_ip}:{port}:{use_ssl}"
         if key not in self.clients:
-            # API REST MikroTik:
+            protocol = "https" if use_ssl or port == 443 else "http"
-            # - Porta 8728: HTTP (default)
-            # - Porta 8729: HTTPS (SSL)
-            protocol = "https" if use_ssl or port == 8729 else "http"
             auth = base64.b64encode(f"{username}:{password}".encode()).decode()
             headers = {
                 "Authorization": f"Basic {auth}",
                 "Content-Type": "application/json"
             }
             
-            # SSL context per MikroTik (supporta protocolli TLS legacy)
             ssl_context = None
             if protocol == "https":
                 ssl_context = ssl.create_default_context()
                 ssl_context.check_hostname = False
                 ssl_context.verify_mode = ssl.CERT_NONE
-                # Abilita protocolli TLS legacy per MikroTik (TLS 1.0+)
                 try:
                     ssl_context.minimum_version = ssl.TLSVersion.TLSv1
                 except AttributeError:
-                    # Python < 3.7 fallback
                     pass
-                # Abilita cipher suite legacy per compatibilità
                 ssl_context.set_ciphers('DEFAULT@SECLEVEL=1')
             
             self.clients[key] = httpx.AsyncClient(
@@ -59,20 +53,41 @@ class MikroTikManager:
             )
         return self.clients[key]
     
-    async def test_connection(self, router_ip: str, username: str, password: str, port: int = 8728, use_ssl: bool = False) -> bool:
+    async def test_connection(self, router_ip: str, username: str, password: str, port: int = 80, use_ssl: bool = False) -> bool:
         """Testa connessione a un router"""
         try:
-            # Auto-detect SSL: porta 8729 = SSL
+            if port == 443:
-            if port == 8729:
                 use_ssl = True
             client = self._get_client(router_ip, username, password, port, use_ssl)
-            # Prova a leggere system identity
             response = await client.get("/rest/system/identity")
             return response.status_code == 200
         except Exception as e:
             print(f"[ERROR] Connessione a {router_ip}:{port} fallita: {e}")
             return False
     
+    async def _get_existing_ips_set(
+        self,
+        router_ip: str,
+        username: str,
+        password: str,
+        list_name: str,
+        port: int = 80,
+        use_ssl: bool = False
+    ) -> Set[str]:
+        """Scarica la address-list UNA VOLTA e ritorna un set di IP già presenti"""
+        try:
+            if port == 443:
+                use_ssl = True
+            client = self._get_client(router_ip, username, password, port, use_ssl)
+            response = await client.get(f"/rest/ip/firewall/address-list", params={"list": list_name})
+            if response.status_code == 200:
+                entries = response.json()
+                return {entry.get('address', '') for entry in entries if entry.get('list') == list_name}
+            return set()
+        except Exception as e:
+            print(f"[ERROR] Lettura address-list da {router_ip}: {e}")
+            return set()
+
     async def add_address_list(
         self,
         router_ip: str,
@@ -82,29 +97,32 @@ class MikroTikManager:
         list_name: str = "ddos_blocked",
         comment: str = "",
         timeout_duration: str = "1h",
-        port: int = 8728,
+        port: int = 80,
-        use_ssl: bool = False
+        use_ssl: bool = False,
+        skip_check: bool = False,
+        existing_ips: Optional[Set[str]] = None
     ) -> bool:
         """
         Aggiunge IP alla address-list del router
-        timeout_duration: es. "1h", "30m", "1d"
+        skip_check: se True, non verifica se l'IP esiste già (per bulk operations)
+        existing_ips: set di IP già nella lista (per evitare GET per ogni IP)
         """
         try:
-            # Auto-detect SSL: porta 8729 = SSL
+            if port == 443:
-            if port == 8729:
                 use_ssl = True
             client = self._get_client(router_ip, username, password, port, use_ssl)
             
-            # Controlla se IP già esiste
+            if not skip_check:
-            response = await client.get("/rest/ip/firewall/address-list")
+                if existing_ips is not None:
-            if response.status_code == 200:
+                    if ip_address in existing_ips:
-                existing = response.json()
-                for entry in existing:
-                    if entry.get('address') == ip_address and entry.get('list') == list_name:
-                        print(f"[INFO] IP {ip_address} già in lista {list_name} su {router_ip}")
                         return True
+                else:
+                    response = await client.get("/rest/ip/firewall/address-list")
+                    if response.status_code == 200:
+                        for entry in response.json():
+                            if entry.get('address') == ip_address and entry.get('list') == list_name:
+                                return True
             
-            # Aggiungi nuovo IP
             data = {
                 "list": list_name,
                 "address": ip_address,
@@ -114,11 +132,25 @@ class MikroTikManager:
             
             response = await client.post("/rest/ip/firewall/address-list/add", json=data)
             
-            if response.status_code == 201 or response.status_code == 200:
+            if response.status_code in (200, 201):
-                print(f"[SUCCESS] IP {ip_address} aggiunto a {list_name} su {router_ip} (timeout: {timeout_duration})")
+                print(f"[SUCCESS] IP {ip_address} aggiunto a {list_name} su {router_ip}")
                 return True
+            elif response.status_code in (400, 409):
+                resp_text = response.text.lower()
+                if "already" in resp_text or "exists" in resp_text or "duplicate" in resp_text or "failure: already" in resp_text:
+                    return True
+                try:
+                    verify_resp = await client.get("/rest/ip/firewall/address-list", params={"address": ip_address})
+                    if verify_resp.status_code == 200:
+                        for entry in verify_resp.json():
+                            if entry.get('address') == ip_address and entry.get('list') == list_name:
+                                return True
+                except Exception:
+                    pass
+                print(f"[ERROR] IP {ip_address} su {router_ip}: {response.status_code} - {response.text}")
+                return False
             else:
-                print(f"[ERROR] Errore aggiunta IP {ip_address} su {router_ip}: {response.status_code} - {response.text}")
+                print(f"[ERROR] Aggiunta IP {ip_address} su {router_ip}: {response.status_code} - {response.text}")
                 return False
                 
         except Exception as e:
@@ -132,17 +164,15 @@ class MikroTikManager:
         password: str,
         ip_address: str,
         list_name: str = "ddos_blocked",
-        port: int = 8728,
+        port: int = 80,
         use_ssl: bool = False
     ) -> bool:
         """Rimuove IP dalla address-list del router"""
         try:
-            # Auto-detect SSL: porta 8729 = SSL
+            if port == 443:
-            if port == 8729:
                 use_ssl = True
             client = self._get_client(router_ip, username, password, port, use_ssl)
             
-            # Trova ID dell'entry
             response = await client.get("/rest/ip/firewall/address-list")
             if response.status_code != 200:
                 return False
@@ -151,7 +181,6 @@ class MikroTikManager:
             for entry in entries:
                 if entry.get('address') == ip_address and entry.get('list') == list_name:
                     entry_id = entry.get('.id')
-                    # Rimuovi entry
                     response = await client.delete(f"/rest/ip/firewall/address-list/{entry_id}")
                     if response.status_code == 200:
                         print(f"[SUCCESS] IP {ip_address} rimosso da {list_name} su {router_ip}")
@@ -170,13 +199,12 @@ class MikroTikManager:
         username: str,
         password: str,
         list_name: Optional[str] = None,
-        port: int = 8728,
+        port: int = 80,
         use_ssl: bool = False
     ) -> List[Dict]:
         """Ottiene address-list da router"""
         try:
-            # Auto-detect SSL: porta 8729 = SSL
+            if port == 443:
-            if port == 8729:
                 use_ssl = True
             client = self._get_client(router_ip, username, password, port, use_ssl)
             response = await client.get("/rest/ip/firewall/address-list")
@@ -203,7 +231,6 @@ class MikroTikManager:
     ) -> Dict[str, bool]:
         """
         Blocca IP su tutti i router in parallelo
-        routers: lista di dict con {ip_address, username, password, api_port}
         Returns: dict con {router_ip: success_bool}
         """
         tasks = []
@@ -221,20 +248,125 @@ class MikroTikManager:
                 list_name=list_name,
                 comment=comment,
                 timeout_duration=timeout_duration,
-                port=router.get('api_port', 8728)
+                port=router.get('api_port', 80)
             )
             tasks.append(task)
             router_ips.append(router['ip_address'])
         
-        # Esegui in parallelo
         results = await asyncio.gather(*tasks, return_exceptions=True)
         
-        # Combina risultati
         return {
             router_ip: result if not isinstance(result, Exception) else False
             for router_ip, result in zip(router_ips, results)
         }
     
+    async def bulk_block_ips_on_all_routers(
+        self,
+        routers: List[Dict],
+        ip_list: List[str],
+        list_name: str = "ddos_blocked",
+        comment_prefix: str = "IDS bulk-block",
+        timeout_duration: str = "1h",
+        concurrency: int = 10,
+        progress_callback=None
+    ) -> Dict[str, Dict[str, bool]]:
+        """
+        Blocco massivo ottimizzato: scarica address-list UNA VOLTA per router,
+        poi aggiunge solo IP non presenti con concurrency limitata.
+        
+        Returns: {ip: {router_ip: success_bool}}
+        """
+        enabled_routers = [r for r in routers if r.get('enabled', True)]
+        if not enabled_routers:
+            return {}
+        
+        print(f"[BULK] Inizio blocco massivo: {len(ip_list)} IP su {len(enabled_routers)} router")
+        
+        existing_cache = {}
+        for router in enabled_routers:
+            router_ip = router['ip_address']
+            port = router.get('api_port', 80)
+            use_ssl = port == 443
+            existing_ips = await self._get_existing_ips_set(
+                router_ip, router['username'], router['password'],
+                list_name, port, use_ssl
+            )
+            existing_cache[router_ip] = existing_ips
+            print(f"[BULK] Router {router_ip}: {len(existing_ips)} IP già in lista")
+        
+        new_ips = []
+        for ip in ip_list:
+            is_new_on_any = False
+            for router in enabled_routers:
+                if ip not in existing_cache.get(router['ip_address'], set()):
+                    is_new_on_any = True
+                    break
+            if is_new_on_any:
+                new_ips.append(ip)
+        
+        already_blocked = len(ip_list) - len(new_ips)
+        print(f"[BULK] {already_blocked} IP già bloccati, {len(new_ips)} nuovi da bloccare")
+        
+        results = {}
+        semaphore = asyncio.Semaphore(concurrency)
+        blocked_count = 0
+        
+        async def block_single_ip(ip: str) -> Dict[str, bool]:
+            nonlocal blocked_count
+            async with semaphore:
+                router_results = {}
+                tasks = []
+                r_ips = []
+                
+                for router in enabled_routers:
+                    r_ip = router['ip_address']
+                    if ip in existing_cache.get(r_ip, set()):
+                        router_results[r_ip] = True
+                        continue
+                    
+                    task = self.add_address_list(
+                        router_ip=r_ip,
+                        username=router['username'],
+                        password=router['password'],
+                        ip_address=ip,
+                        list_name=list_name,
+                        comment=f"{comment_prefix} {ip}",
+                        timeout_duration=timeout_duration,
+                        port=router.get('api_port', 80),
+                        skip_check=True
+                    )
+                    tasks.append(task)
+                    r_ips.append(r_ip)
+                
+                if tasks:
+                    task_results = await asyncio.gather(*tasks, return_exceptions=True)
+                    for r_ip, result in zip(r_ips, task_results):
+                        router_results[r_ip] = result if not isinstance(result, Exception) else False
+                
+                blocked_count += 1
+                if progress_callback and blocked_count % 50 == 0:
+                    await progress_callback(blocked_count, len(new_ips))
+                
+                return router_results
+        
+        batch_tasks = [block_single_ip(ip) for ip in new_ips]
+        batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
+        
+        for ip, result in zip(new_ips, batch_results):
+            if isinstance(result, Exception):
+                results[ip] = {r['ip_address']: False for r in enabled_routers}
+            else:
+                results[ip] = result
+        
+        for ip in ip_list:
+            if ip not in results:
+                results[ip] = {r['ip_address']: True for r in enabled_routers}
+        
+        total_success = sum(1 for ip_results in results.values() if any(ip_results.values()))
+        print(f"[BULK] Completato: {total_success}/{len(ip_list)} IP bloccati con successo")
+        
+        return results
+    
     async def unblock_ip_on_all_routers(
         self,
         routers: List[Dict],
@@ -255,7 +387,7 @@ class MikroTikManager:
                 password=router['password'],
                 ip_address=ip_address,
                 list_name=list_name,
-                port=router.get('api_port', 8728)
+                port=router.get('api_port', 80)
             )
             tasks.append(task)
             router_ips.append(router['ip_address'])
@@ -274,7 +406,6 @@ class MikroTikManager:
         self.clients.clear()
 
 
-# Fallback SSH per router che non supportano API REST
 class MikroTikSSHManager:
     """Fallback usando SSH se API REST non disponibile"""
     
@@ -288,29 +419,26 @@ class MikroTikSSHManager:
 
 
 if __name__ == "__main__":
-    # Test MikroTik Manager
     async def test():
         manager = MikroTikManager()
         
-        # Test router demo (sostituire con dati reali)
         test_router = {
             'ip_address': '192.168.1.1',
             'username': 'admin',
             'password': 'password',
-            'api_port': 8728,
+            'api_port': 80,
             'enabled': True
         }
         
-        # Test connessione
         print("Testing connection...")
         connected = await manager.test_connection(
             test_router['ip_address'],
             test_router['username'],
-            test_router['password']
+            test_router['password'],
+            port=test_router['api_port']
         )
         print(f"Connected: {connected}")
         
-        # Test blocco IP
         if connected:
             print("\nTesting IP block...")
             result = await manager.add_address_list(
@@ -320,22 +448,22 @@ if __name__ == "__main__":
                 ip_address='10.0.0.100',
                 list_name='ddos_test',
                 comment='Test IDS',
-                timeout_duration='10m'
+                timeout_duration='10m',
+                port=test_router['api_port']
             )
             print(f"Block result: {result}")
             
-            # Leggi lista
             print("\nReading address list...")
             entries = await manager.get_address_list(
                 test_router['ip_address'],
                 test_router['username'],
                 test_router['password'],
-                list_name='ddos_test'
+                list_name='ddos_test',
+                port=test_router['api_port']
             )
             print(f"Entries: {entries}")
         
         await manager.close_all()
     
-    # Esegui test
     print("=== TEST MIKROTIK MANAGER ===\n")
     asyncio.run(test())

server/routes.ts

@@ -492,10 +492,16 @@ export async function registerRoutes(app: Express): Promise<Server> {
         high: sql<number>`count(*) filter (where ${detections.riskScore}::numeric >= 70 and ${detections.riskScore}::numeric < 85)::int`,
       }).from(detections);
 
-      const logStats = await db.select({
+      let logCount = 0;
-        count: sql<number>`count(*)::int`,
+      try {
-      }).from(networkLogs)
+        const logStats = await db.execute(
-        .where(gte(networkLogs.timestamp, new Date(Date.now() - 24 * 60 * 60 * 1000)));
+          sql`SELECT count(*)::int as count FROM network_logs WHERE timestamp >= NOW() - INTERVAL '24 hours'`
+        );
+        logCount = (logStats as any).rows?.[0]?.count ?? (logStats as any)[0]?.count ?? 0;
+      } catch (logError) {
+        console.error('[DB WARN] Log count query failed:', logError);
+        logCount = 0;
+      }
 
       res.json({
         routers: {
@@ -509,7 +515,7 @@ export async function registerRoutes(app: Express): Promise<Server> {
           high: detectionStats[0]?.high || 0
         },
         logs: {
-          recent: logStats[0]?.count || 0
+          recent: logCount
         },
         whitelist: {
           total: whitelistResult.total