(self, capture_directory)
| 303 | class Engine(): |
| 304 | |
| 305 | def __init__(self, capture_directory): |
| 306 | |
| 307 | # Set some vars. |
| 308 | self.analysis_start = datetime.now() |
| 309 | self.connected = self.check_internet() |
| 310 | self.working_dir = capture_directory |
| 311 | self.assets_dir = f"{capture_directory}/assets/" |
| 312 | self.rules_file = "/tmp/rules.rules" |
| 313 | self.pcap_path = os.path.join(self.working_dir, "capture.pcap") |
| 314 | self.records = [] |
| 315 | self.alerts = [] |
| 316 | self.dns = [] |
| 317 | self.files = [] |
| 318 | self.whitelist = [] |
| 319 | self.uncategorized = [] |
| 320 | self.analysed = [] |
| 321 | self.dns_failed = [] |
| 322 | self.dns_checked = set() |
| 323 | self.cert_checked = set() |
| 324 | self.errors = [] |
| 325 | self.analysis_end = None |
| 326 | self._enabled_indicator_types = None |
| 327 | self._active_ssl_cache = {} |
| 328 | self._active_ssl_lock = threading.Lock() |
| 329 | self._active_ssl_workers = 4 |
| 330 | # Hostname extracted from peer TLS certificate (CN/SAN), keyed by (host_or_ip, port). |
| 331 | self._tls_cert_hostname_cache: dict[tuple[str, int], Optional[str]] = {} |
| 332 | # Unique local IPv6 (fc00::/7); used instead of a broken list-comprehension test. |
| 333 | self._ipv6_ula = IPv6Network("fc00::/7") |
| 334 | self._umbrella_top500k_set: Optional[set[str]] = None |
| 335 | self._umbrella_top500k_rank_map: Optional[dict[str, int]] = None |
| 336 | self._ipthc_cache: dict[str, tuple[Optional[str], int]] = {} |
| 337 | # Domain enrichment caches (filled via prefetch to avoid sequential network latency). |
| 338 | self._domain_ns_cache: dict[str, list[str]] = {} |
| 339 | self._domain_ns_err: dict[str, str] = {} |
| 340 | self._domain_whois_creation_cache: dict[str, Optional[datetime]] = {} |
| 341 | self._domain_whois_err: dict[str, str] = {} |
| 342 | self._domain_enrich_lock = threading.Lock() |
| 343 | self._enrich_workers = 4 |
| 344 | |
| 345 | # Get configuration (centralized/typed) |
| 346 | self.config = EngineConfig.load(get_config) |
| 347 | self.heuristics_analysis = self.config.heuristics_analysis |
| 348 | self.iocs_analysis = self.config.iocs_analysis |
| 349 | self.whitelist_analysis = self.config.whitelist_analysis |
| 350 | self.active_analysis = self.config.active_analysis |
| 351 | self.userlang = self.config.userlang |
| 352 | self.max_ports = self.config.max_ports |
| 353 | self.http_default_ports = self.config.http_default_ports |
| 354 | self.tls_default_ports = self.config.tls_default_ports |
| 355 | self.indicators_types = self.config.indicators_types |
| 356 | |
| 357 | # Save detection methods used. |
| 358 | self.detection_methods = { |
| 359 | "iocs": self.iocs_analysis, |
| 360 | "heuristics": self.heuristics_analysis, |
| 361 | "active": self.active_analysis, |
| 362 | } |
nothing calls this directly
no test coverage detected