# File: src/reconnaissance.py """Main reconnaissance logic.""" import threading import concurrent.futures import logging from datetime import datetime from typing import Set, List, Optional from .data_structures import ReconData from .config import Config from .dns_resolver import DNSResolver from .certificate_checker import CertificateChecker from .shodan_client import ShodanClient from .virustotal_client import VirusTotalClient from .tld_fetcher import TLDFetcher # Set up logging for this module logger = logging.getLogger(__name__) class ReconnaissanceEngine: """Main reconnaissance engine.""" def __init__(self, config: Config): self.config = config # Initialize clients self.dns_resolver = DNSResolver(config) self.cert_checker = CertificateChecker(config) self.tld_fetcher = TLDFetcher() # Optional clients self.shodan_client = None if config.shodan_key: self.shodan_client = ShodanClient(config.shodan_key, config) logger.info("✅ Shodan client initialized") else: logger.info("⚠️ Shodan API key not provided, skipping Shodan integration") self.virustotal_client = None if config.virustotal_key: self.virustotal_client = VirusTotalClient(config.virustotal_key, config) logger.info("✅ VirusTotal client initialized") else: logger.info("⚠️ VirusTotal API key not provided, skipping VirusTotal integration") # Progress tracking self.progress_callback = None self._lock = threading.Lock() # Shared data object for live updates self.shared_data = None def set_progress_callback(self, callback): """Set callback for progress updates.""" self.progress_callback = callback def set_shared_data(self, shared_data: ReconData): """Set shared data object for live updates during web interface usage.""" self.shared_data = shared_data logger.info("📊 Using shared data object for live updates") def _update_progress(self, message: str, percentage: int = None): """Update progress if callback is set.""" logger.info(f"Progress: {message} ({percentage}%)" if percentage else f"Progress: {message}") if self.progress_callback: self.progress_callback(message, percentage) def run_reconnaissance(self, target: str) -> ReconData: """Run full reconnaissance on target.""" # Use shared data object if available, otherwise create new one if self.shared_data is not None: self.data = self.shared_data logger.info("📊 Using shared data object for reconnaissance") else: self.data = ReconData() logger.info("📊 Created new data object for reconnaissance") self.data.start_time = datetime.now() logger.info(f"🚀 Starting reconnaissance for target: {target}") logger.info(f"📊 Configuration: max_depth={self.config.max_depth}, " f"DNS_rate={self.config.DNS_RATE_LIMIT}/s") try: # Determine if target is hostname.tld or just hostname if '.' in target: logger.info(f"🎯 Target '{target}' appears to be a full domain name") self._update_progress(f"Starting reconnaissance for {target}", 0) self.data.add_hostname(target, 0) initial_targets = {target} else: logger.info(f"🔍 Target '{target}' appears to be a hostname, expanding to all TLDs") self._update_progress(f"Expanding {target} to all TLDs", 5) initial_targets = self._expand_hostname_to_tlds(target) logger.info(f"📋 Found {len(initial_targets)} valid domains after TLD expansion") self._update_progress("Resolving initial targets", 10) # Process all targets recursively self._process_targets_recursively(initial_targets) # Final external lookups self._update_progress("Performing external service lookups", 90) self._perform_external_lookups() # Log final statistics stats = self.data.get_stats() logger.info(f"📈 Final statistics: {stats}") self._update_progress("Reconnaissance complete", 100) except Exception as e: logger.error(f"❌ Error during reconnaissance: {e}", exc_info=True) raise finally: self.data.end_time = datetime.now() duration = self.data.end_time - self.data.start_time logger.info(f"⏱️ Total reconnaissance time: {duration}") return self.data def _expand_hostname_to_tlds(self, hostname: str) -> Set[str]: """Expand hostname to all possible TLDs.""" logger.info(f"🌐 Fetching TLD list for hostname expansion") tlds = self.tld_fetcher.get_tlds() logger.info(f"🔍 Testing against {len(tlds)} TLDs") targets = set() tested_count = 0 for i, tld in enumerate(tlds): full_hostname = f"{hostname}.{tld}" # Quick check if domain resolves ips = self.dns_resolver.resolve_hostname(full_hostname) tested_count += 1 if ips: logger.info(f"✅ Found valid domain: {full_hostname} -> {ips}") self.data.add_hostname(full_hostname, 0) targets.add(full_hostname) for ip in ips: self.data.add_ip_address(ip) # Progress update every 100 TLDs if i % 100 == 0: progress = 5 + int((i / len(tlds)) * 5) # 5-10% range self._update_progress(f"Checked {i}/{len(tlds)} TLDs, found {len(targets)} valid domains", progress) logger.info(f"🎯 TLD expansion complete: tested {tested_count} TLDs, found {len(targets)} valid domains") return targets def _process_targets_recursively(self, targets: Set[str]): """Process targets with recursive subdomain discovery.""" current_depth = 0 while current_depth <= self.config.max_depth and targets: logger.info(f"🔄 Processing depth {current_depth} with {len(targets)} targets") self._update_progress(f"Processing depth {current_depth} ({len(targets)} targets)", 15 + (current_depth * 25)) new_targets = set() for target in targets: logger.debug(f"🔍 Processing target: {target}") # DNS resolution and record gathering self._process_single_target(target, current_depth) # Extract new subdomains if current_depth < self.config.max_depth: new_subdomains = self._extract_new_subdomains(target) logger.debug(f"🌿 Found {len(new_subdomains)} new subdomains from {target}") for subdomain in new_subdomains: self.data.add_hostname(subdomain, current_depth + 1) new_targets.add(subdomain) logger.info(f"📊 Depth {current_depth} complete. Found {len(new_targets)} new targets for next depth") targets = new_targets current_depth += 1 logger.info(f"🏁 Recursive processing complete after {current_depth} levels") def _process_single_target(self, hostname: str, depth: int): """Process a single target hostname.""" logger.debug(f"🎯 Processing single target: {hostname} at depth {depth}") # Get all DNS records dns_records = self.dns_resolver.get_all_dns_records(hostname) logger.debug(f"📋 Found {len(dns_records)} DNS records for {hostname}") for record in dns_records: self.data.add_dns_record(hostname, record) # Extract IP addresses from A and AAAA records if record.record_type in ['A', 'AAAA']: self.data.add_ip_address(record.value) # Get certificates logger.debug(f"🔍 Checking certificates for {hostname}") certificates = self.cert_checker.get_certificates(hostname) if certificates: self.data.certificates[hostname] = certificates logger.info(f"📜 Found {len(certificates)} certificates for {hostname}") else: logger.debug(f"❌ No certificates found for {hostname}") def _extract_new_subdomains(self, hostname: str) -> Set[str]: """Extract new subdomains from DNS records and certificates.""" new_subdomains = set() # From DNS records if hostname in self.data.dns_records: dns_subdomains = self.dns_resolver.extract_subdomains_from_dns( self.data.dns_records[hostname] ) new_subdomains.update(dns_subdomains) logger.debug(f"🌐 Extracted {len(dns_subdomains)} subdomains from DNS records of {hostname}") # From certificates if hostname in self.data.certificates: cert_subdomains = self.cert_checker.extract_subdomains_from_certificates( self.data.certificates[hostname] ) new_subdomains.update(cert_subdomains) logger.debug(f"🔐 Extracted {len(cert_subdomains)} subdomains from certificates of {hostname}") # Filter out already known hostnames filtered_subdomains = new_subdomains - self.data.hostnames logger.debug(f"🆕 {len(filtered_subdomains)} new subdomains after filtering") return filtered_subdomains def _perform_external_lookups(self): """Perform Shodan and VirusTotal lookups.""" logger.info(f"🔍 Starting external lookups for {len(self.data.ip_addresses)} IPs and {len(self.data.hostnames)} hostnames") # Reverse DNS for all IPs logger.info("🔄 Performing reverse DNS lookups") reverse_dns_count = 0 for ip in self.data.ip_addresses: reverse = self.dns_resolver.reverse_dns_lookup(ip) if reverse: self.data.reverse_dns[ip] = reverse reverse_dns_count += 1 logger.debug(f"🔙 Reverse DNS for {ip}: {reverse}") logger.info(f"✅ Completed reverse DNS: {reverse_dns_count}/{len(self.data.ip_addresses)} successful") # Shodan lookups if self.shodan_client: logger.info(f"🕵️ Starting Shodan lookups for {len(self.data.ip_addresses)} IPs") shodan_success_count = 0 for ip in self.data.ip_addresses: try: logger.debug(f"🔍 Querying Shodan for IP: {ip}") result = self.shodan_client.lookup_ip(ip) if result: self.data.add_shodan_result(ip, result) shodan_success_count += 1 logger.info(f"✅ Shodan result for {ip}: {len(result.ports)} ports") else: logger.debug(f"❌ No Shodan data for {ip}") except Exception as e: logger.warning(f"⚠️ Error querying Shodan for {ip}: {e}") logger.info(f"✅ Shodan lookups complete: {shodan_success_count}/{len(self.data.ip_addresses)} successful") else: logger.info("⚠️ Skipping Shodan lookups (no API key)") # VirusTotal lookups if self.virustotal_client: total_resources = len(self.data.ip_addresses) + len(self.data.hostnames) logger.info(f"🛡️ Starting VirusTotal lookups for {total_resources} resources") vt_success_count = 0 # Check IPs for ip in self.data.ip_addresses: try: logger.debug(f"🔍 Querying VirusTotal for IP: {ip}") result = self.virustotal_client.lookup_ip(ip) if result: self.data.add_virustotal_result(ip, result) vt_success_count += 1 logger.info(f"🛡️ VirusTotal result for {ip}: {result.positives}/{result.total} detections") else: logger.debug(f"❌ No VirusTotal data for {ip}") except Exception as e: logger.warning(f"⚠️ Error querying VirusTotal for IP {ip}: {e}") # Check domains for hostname in self.data.hostnames: try: logger.debug(f"🔍 Querying VirusTotal for domain: {hostname}") result = self.virustotal_client.lookup_domain(hostname) if result: self.data.add_virustotal_result(hostname, result) vt_success_count += 1 logger.info(f"🛡️ VirusTotal result for {hostname}: {result.positives}/{result.total} detections") else: logger.debug(f"❌ No VirusTotal data for {hostname}") except Exception as e: logger.warning(f"⚠️ Error querying VirusTotal for domain {hostname}: {e}") logger.info(f"✅ VirusTotal lookups complete: {vt_success_count}/{total_resources} successful") else: logger.info("⚠️ Skipping VirusTotal lookups (no API key)") # Final external lookup summary ext_stats = { 'reverse_dns': len(self.data.reverse_dns), 'shodan_results': len(self.data.shodan_results), 'virustotal_results': len(self.data.virustotal_results) } logger.info(f"📊 External lookups summary: {ext_stats}")