Stabilize UTC timestamps and dev tooling

Author: XploitGost

backend/app/main.py

@@ -3,6 +3,8 @@
 import csv
 import io
 import logging
+import os
+import sys
 import time
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
@@ -103,8 +105,11 @@
 saved_filter_service = SavedFilterService()
 logger = logging.getLogger("netbotpro.api")
 if not logger.handlers:
+    default_level = "CRITICAL" if "unittest" in sys.modules else "INFO"
+    log_level = os.environ.get("NETBOT_LOG_LEVEL", default_level).upper()
     logging.basicConfig(
-        level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s"
+        level=getattr(logging, log_level, logging.INFO),
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
     )
 MAX_PCAP_UPLOAD_BYTES = 50 * 1024 * 1024
 ALLOWED_PCAP_SUFFIXES = {".pcap", ".pcapng"}

backend/app/services/sniffer_service.py

@@ -4,7 +4,7 @@
 import os
 import queue
 import threading
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any
 
 from backend.app.bootstrap import ensure_project_root_on_path
@@ -189,7 +189,7 @@ def _on_packet(self, meta: dict[str, Any]) -> None:
 
     def _process_packet(self, meta: dict[str, Any]) -> None:
         packet = dict(meta)
-        packet.setdefault("ts", datetime.utcnow().isoformat() + "Z")
+        packet.setdefault("ts", datetime.now(timezone.utc).isoformat())
         packet.setdefault("id", self._next_packet_id())
         policy = current_capture_policy()
         self._apply_payload_policy(packet, policy.to_public_dict())

core/ids_rules_engine.py

@@ -1,10 +1,10 @@
 # ids_rules_engine.py
 # -*- coding: utf-8 -*-
-import os
 import json
+import os
 from collections import defaultdict
-from datetime import datetime, timedelta
-from typing import Dict, Any, List, Tuple
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Tuple
 
 
 class RuleEngine:
@@ -66,7 +66,7 @@ def analyze(self, meta: Dict[str, Any]) -> Dict[str, Any] | None:
 
         src = meta.get("src")
         dport = int(meta.get("dport") or 0)
-        now = datetime.utcnow()
+        now = datetime.now(timezone.utc)
 
         for idx, rule in enumerate(self.rules):
             if not rule or not rule.get("enabled", True):

core/ids_signature.py

@@ -12,11 +12,11 @@
 
 from __future__ import annotations
 
-from collections import defaultdict
-from datetime import datetime, timedelta
-from typing import Dict, Any, List, Tuple, Optional
 import json
 import os
+from collections import defaultdict
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Optional, Tuple
 
 
 class SignatureIDS:
@@ -28,7 +28,9 @@ class SignatureIDS:
 
     def __init__(self) -> None:
         # (src, dst) -> list[(ts, dport)]
-        self.port_scan_map: Dict[Tuple[str, str], List[Tuple[datetime, int]]] = defaultdict(list)
+        self.port_scan_map: Dict[Tuple[str, str], List[Tuple[datetime, int]]] = (
+            defaultdict(list)
+        )
         # (src, dst) -> list[ts]
         self.syn_counter: Dict[Tuple[str, str], List[datetime]] = defaultdict(list)
         # src -> list[ts]
@@ -47,7 +49,9 @@ def __init__(self) -> None:
         # Custom rules for Rule Editor
         self.custom_rules: List[Dict[str, Any]] = []
         # (rule_idx, src, dport) -> list[datetime]
-        self.custom_state: Dict[Tuple[int, str, int], List[datetime]] = defaultdict(list)
+        self.custom_state: Dict[Tuple[int, str, int], List[datetime]] = defaultdict(
+            list
+        )
 
         base_dir = os.path.abspath(os.path.dirname(__file__))
         self.rules_path: str = os.path.join(base_dir, "rules.json")
@@ -149,15 +153,17 @@ def analyze_packet(self, meta: Dict[str, Any]) -> Optional[Dict[str, Any]]:
         Returns:
           alert dict or None
         """
-        now = datetime.utcnow()
+        now = datetime.now(timezone.utc)
         self._cleanup(now)
 
         src = meta.get("src")
         dst = meta.get("dst")
         proto = meta.get("proto")
         dport = meta.get("dport")
-        flags = (meta.get("flags") or "")
-        length = int(meta.get("length") or 0)  # فعلاً استفاده نمی‌کنیم ولی برای آینده خوبه
+        flags = meta.get("flags") or ""
+        length = int(
+            meta.get("length") or 0
+        )  # فعلاً استفاده نمی‌کنیم ولی برای آینده خوبه
 
         alert: Optional[Dict[str, Any]] = None
 
@@ -176,7 +182,13 @@ def analyze_packet(self, meta: Dict[str, Any]) -> Optional[Dict[str, Any]]:
                 }
 
         # ---------------- Port Scan detection ----------------
-        if alert is None and src and dst and dport is not None and proto in ("TCP", "UDP"):
+        if (
+            alert is None
+            and src
+            and dst
+            and dport is not None
+            and proto in ("TCP", "UDP")
+        ):
             key = (src, dst)
             arr2 = self.port_scan_map[key]
             arr2.append((now, int(dport)))

core/incident_engine.py

@@ -14,7 +14,7 @@
 
 import hashlib
 from dataclasses import dataclass
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any, Dict, Optional, Tuple
 
 
@@ -47,21 +47,25 @@ def __init__(self, window_sec: int = 120) -> None:
         self.window = timedelta(seconds=int(window_sec))
         self._incidents: Dict[Tuple[str, str, str, str, str], Incident] = {}
 
-    def _make_key(self, meta: Dict[str, Any], alert: Dict[str, Any]) -> Tuple[str, str, str, str, str]:
+    def _make_key(
+        self, meta: Dict[str, Any], alert: Dict[str, Any]
+    ) -> Tuple[str, str, str, str, str]:
         src = _safe_str(meta.get("src"))
         dst = _safe_str(meta.get("dst"))
         atk = _safe_str(alert.get("attack_type") or alert.get("attack") or "Alert")
         dport = _safe_str(meta.get("dport"))
         eng = _safe_str(alert.get("engine") or "IDS")
         return (src, dst, atk, dport, eng)
 
-    def _make_id(self, key: Tuple[str, str, str, str, str], first_seen: datetime) -> str:
+    def _make_id(
+        self, key: Tuple[str, str, str, str, str], first_seen: datetime
+    ) -> str:
         base = "|".join(key) + "|" + first_seen.strftime("%Y%m%d%H%M%S")
         h = hashlib.sha1(base.encode("utf-8", errors="ignore")).hexdigest()
         return h[:12]
 
     def enrich_alert(self, meta: Dict[str, Any], alert: Dict[str, Any]) -> None:
-        now = datetime.utcnow()
+        now = datetime.now(timezone.utc)
         key = self._make_key(meta, alert)
 
         # Parse normalized score if present