mstoeck3/timesketch_analyzers
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

whois_analyzer.py aktualisiert

Browse Source
This commit is contained in:
Mario Stöckl 2025-08-25 19:56:38 +00:00
parent bcbe7a8bd7
commit 90f915c989
1 changed files with 8 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
whois_analyzer.py
View File

@ -34,7 +34,6 @@ class WhoisEnrichmentAnalyzer(interface.BaseAnalyzer):
self.timeout = current_app.config.get('WHOIS_TIMEOUT', 10) self.timeout = current_app.config.get('WHOIS_TIMEOUT', 10)
self.whois_cache: Dict[str, Optional[Dict]] = {} self.whois_cache: Dict[str, Optional[Dict]] = {}
self.processed_ips: Set[str] = set()
def _validate_ip(self, ip_address: str) -> bool: def _validate_ip(self, ip_address: str) -> bool:
"""Validate IP address.""" """Validate IP address."""
@ -168,7 +167,7 @@ class WhoisEnrichmentAnalyzer(interface.BaseAnalyzer):
for event in events: for event in events:
events_processed += 1 events_processed += 1
# Find first valid IP in this event # Find first valid IP in this event and enrich it
for ip_field in self.IP_FIELDS: for ip_field in self.IP_FIELDS:
ip_value = event.source.get(ip_field) ip_value = event.source.get(ip_field)
if not ip_value: if not ip_value:
@ -183,15 +182,13 @@ class WhoisEnrichmentAnalyzer(interface.BaseAnalyzer):
if not self._validate_ip(ip_str): if not self._validate_ip(ip_str):
continue continue
if ip_str not in self.processed_ips: # Process EVERY IP, no duplicate checking
self.processed_ips.add(ip_str) whois_data = self._get_whois_data(ip_str) # Uses cache to avoid duplicate API calls
whois_data = self._get_whois_data(ip_str) if whois_data:
self._enrich_event(event, ip_field, ip_str, whois_data)
if whois_data: enriched_count += 1
self._enrich_event(event, ip_field, ip_str, whois_data) break # Only enrich first valid IP per event
enriched_count += 1
break
# Rate limiting # Rate limiting
if events_processed % self.batch_size == 0: if events_processed % self.batch_size == 0: