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

shodan_analyzer.py aktualisiert

Browse Source
This commit is contained in:
Mario Stöckl 2025-08-25 12:45:48 +00:00
parent 4fd7bc0fbc
commit db302f4a55
1 changed files with 46 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

87
shodan_analyzer.py
View File

@ -4,6 +4,7 @@ import requests
import json import json
from datetime import datetime from datetime import datetime
import ipaddress import ipaddress
import os
class ShodanEnrichmentAnalyzer(interface.BaseAnalyzer): class ShodanEnrichmentAnalyzer(interface.BaseAnalyzer):
"""Analyzer to enrich IP addresses with Shodan data.""" """Analyzer to enrich IP addresses with Shodan data."""
@ -15,14 +16,13 @@ class ShodanEnrichmentAnalyzer(interface.BaseAnalyzer):
def __init__(self, index_name, sketch_id, timeline_id=None): def __init__(self, index_name, sketch_id, timeline_id=None):
super().__init__(index_name, sketch_id, timeline_id) super().__init__(index_name, sketch_id, timeline_id)
# Get API key from config or environment # Get API key from environment variables
config = self.get_config() self.shodan_api_key = os.environ.get('SHODAN_API_KEY', '')
self.shodan_api_key = config.get('api_key', '') self.max_time_diff_hours = 24
self.max_time_diff_hours = config.get('max_time_diff_hours', 24) self.rate_limit_delay = 1
self.rate_limit_delay = config.get('rate_limit_delay', 1)
if not self.shodan_api_key: if not self.shodan_api_key:
self.logger.error("Shodan API key not configured") self.logger.error("Shodan API key not configured in environment variables")
def run(self): def run(self):
"""Main analyzer logic.""" """Main analyzer logic."""
@ -48,6 +48,7 @@ class ShodanEnrichmentAnalyzer(interface.BaseAnalyzer):
timestamp = event.source.get('timestamp') timestamp = event.source.get('timestamp')
if source_ip and self._is_public_ip(source_ip): if source_ip and self._is_public_ip(source_ip):
self.logger.info(f"Processing IP: {source_ip}")
shodan_data = self._get_shodan_data(source_ip) shodan_data = self._get_shodan_data(source_ip)
if shodan_data: if shodan_data:
self._enrich_event(event, shodan_data) self._enrich_event(event, shodan_data)
@ -68,55 +69,58 @@ class ShodanEnrichmentAnalyzer(interface.BaseAnalyzer):
'history': 'true' 'history': 'true'
} }
self.logger.info(f"Querying Shodan API for IP: {ip}")
response = requests.get(url, params=params, timeout=10) response = requests.get(url, params=params, timeout=10)
if response.status_code == 200: if response.status_code == 200:
self.logger.info(f"Successfully retrieved Shodan data for {ip}")
return response.json() return response.json()
elif response.status_code == 404: elif response.status_code == 404:
self.logger.debug(f'No Shodan data found for {ip}') self.logger.debug(f'No Shodan data found for {ip}')
return None return None
else: else:
self.logger.warning(f'Shodan API error for {ip}: {response.status_code}') self.logger.warning(f'Shodan API error for {ip}: {response.status_code} - {response.text}')
return None return None
except requests.exceptions.RequestException as e: except Exception as e:
self.logger.warning(f'Request error for {ip}: {e}') self.logger.warning(f'Error fetching Shodan data for {ip}: {e}')
return None
except json.JSONDecodeError as e:
self.logger.warning(f'JSON decode error for {ip}: {e}')
return None return None
def _enrich_event(self, event, shodan_data): def _enrich_event(self, event, shodan_data):
"""Add Shodan data to the event.""" """Add Shodan data to the event."""
enrichment = { try:
'shodan_ports': shodan_data.get('ports', []), enrichment = {
'shodan_org': shodan_data.get('org', ''), 'shodan_ports': shodan_data.get('ports', []),
'shodan_isp': shodan_data.get('isp', ''), 'shodan_org': shodan_data.get('org', ''),
'shodan_country': shodan_data.get('location', {}).get('country_name', ''), 'shodan_isp': shodan_data.get('isp', ''),
'shodan_city': shodan_data.get('location', {}).get('city', ''), 'shodan_country': shodan_data.get('location', {}).get('country_name', ''),
'shodan_hostnames': shodan_data.get('hostnames', []), 'shodan_city': shodan_data.get('location', {}).get('city', ''),
'shodan_last_update': shodan_data.get('last_update', ''), 'shodan_hostnames': shodan_data.get('hostnames', []),
'__ts_analyzer_shodan_enrichment': True 'shodan_last_update': shodan_data.get('last_update', ''),
} '__ts_analyzer_shodan_enrichment': True
}
# Add service banners from latest scan
if shodan_data.get('data'):
latest_scan = shodan_data['data'][0] # Most recent scan
enrichment['shodan_services'] = []
for service in shodan_data.get('data', []): # Add service information from latest scan
service_info = { if shodan_data.get('data'):
'port': service.get('port'), services = []
'protocol': service.get('transport', 'tcp'), for service in shodan_data.get('data', [])[:5]: # Limit to first 5 services
'service': service.get('product', ''), service_info = f"Port {service.get('port', 'Unknown')}/{service.get('transport', 'tcp')}"
'version': service.get('version', ''), if service.get('product'):
'timestamp': service.get('timestamp', '') service_info += f" - {service.get('product', '')}"
} if service.get('version'):
enrichment['shodan_services'].append(service_info) service_info += f" {service.get('version', '')}"
services.append(service_info)
event.add_attributes(enrichment)
event.add_tags(['shodan-enriched']) enrichment['shodan_services'] = services
event.commit()
event.add_attributes(enrichment)
event.add_tags(['shodan-enriched'])
event.commit()
self.logger.info(f"Successfully enriched event with Shodan data")
except Exception as e:
self.logger.error(f"Error enriching event: {e}")
def _is_public_ip(self, ip): def _is_public_ip(self, ip):
"""Check if IP is public (not RFC1918 private ranges).""" """Check if IP is public (not RFC1918 private ranges)."""
@ -124,7 +128,8 @@ class ShodanEnrichmentAnalyzer(interface.BaseAnalyzer):
ip_obj = ipaddress.ip_address(ip) ip_obj = ipaddress.ip_address(ip)
return ip_obj.is_global return ip_obj.is_global
except (ValueError, ipaddress.AddressValueError): except (ValueError, ipaddress.AddressValueError):
self.logger.debug(f"Invalid IP address format: {ip}")
return False return False
# Register the analyzer # Register the analyzer
manager.AnalysisManager.register_analyzer(ShodanEnrichmentAnalyzer) manager.AnalysisManager.register_analyzer(ShodanEnrichmentAnalyzer)