shadowhunter_tor.py

#!/usr/bin/env python3
"""
ShadowHunter - Dark Web Credential Intelligence Platform
Module: Tor Network Integration & Dark Web Scraping
Author: Fevra
Version: 0.1.0

LEGAL NOTICE:
This tool is for AUTHORIZED security research and threat intelligence only.
Only access dark web sites where you have explicit permission or for legitimate
threat intelligence gathering. Always comply with applicable laws.
"""

import asyncio
import aiohttp
import socket
try:
    import socks
    SOCKS_AVAILABLE = True
except ImportError:
    socks = None
    SOCKS_AVAILABLE = False
try:
    import stem
    from stem import Signal
    from stem.control import Controller
    STEM_AVAILABLE = True
except ImportError:
    stem = None
    Signal = None
    Controller = None
    STEM_AVAILABLE = False
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
import logging
import time
import random
import re
from urllib.parse import urljoin, urlparse
import hashlib

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('ShadowHunter.Tor')


@dataclass
class OnionSite:
    """Dark web site structure"""
    url: str
    name: str
    category: str
    last_checked: Optional[str] = None
    is_online: bool = False
    title: Optional[str] = None
    
    def get_domain(self) -> str:
        """Extract .onion domain"""
        parsed = urlparse(self.url)
        return parsed.netloc


@dataclass
class DarkWebListing:
    """Dark web marketplace/forum listing"""
    title: str
    description: str
    price: Optional[float]
    seller: str
    category: str
    source_url: str
    posted_date: Optional[str]
    data_type: str  # credentials, access, exploit, etc.
    raw_html: str = ""
    extracted_data: Dict = None
    
    def __post_init__(self):
        if self.extracted_data is None:
            self.extracted_data = {}


class TorConnection:
    """Manage Tor connection and circuits"""
    
    def __init__(self, 
                 tor_proxy_host: str = '127.0.0.1',
                 tor_proxy_port: int = 9050,
                 tor_control_port: int = 9051,
                 tor_password: Optional[str] = None):
        """
        Initialize Tor connection
        
        Args:
            tor_proxy_host: SOCKS proxy host
            tor_proxy_port: SOCKS proxy port (default 9050)
            tor_control_port: Control port (default 9051)
            tor_password: Tor control password (if set)
        """
        self.proxy_host = tor_proxy_host
        self.proxy_port = tor_proxy_port
        self.control_port = tor_control_port
        self.tor_password = tor_password
        self.controller = None
        
        logger.info(f"Tor proxy configured: {tor_proxy_host}:{tor_proxy_port}")
    
    async def connect(self) -> bool:
        """Establish Tor connection. SOCKS and control port require PySocks and stem."""
        try:
            # Test SOCKS proxy (no-op if PySocks not installed)
            self._test_socks_connection()
            
            # Connect to control port (optional; requires stem)
            if STEM_AVAILABLE and Controller is not None:
                self.controller = Controller.from_port(port=self.control_port)
                if self.tor_password:
                    self.controller.authenticate(password=self.tor_password)
                else:
                    self.controller.authenticate()
                logger.info("✓ Connected to Tor network")
            else:
                logger.debug("stem not installed; Tor control port not used. Install with: pip install stem")
            return True
            
        except Exception as e:
            logger.error(f"Failed to connect to Tor: {e}")
            logger.error("Make sure Tor is running: 'tor' or 'brew services start tor'")
            return False
    
    def _test_socks_connection(self):
        """Test SOCKS proxy connectivity. No-op if PySocks not installed."""
        if not SOCKS_AVAILABLE or socks is None:
            logger.warning("PySocks not installed; Tor SOCKS proxy will not be used. Install with: pip install PySocks")
            return
        try:
            socks.set_default_proxy(socks.SOCKS5, self.proxy_host, self.proxy_port)
            socket.socket = socks.socksocket
            logger.info("✓ SOCKS proxy connection successful")
        except Exception as e:
            raise ConnectionError(f"SOCKS proxy test failed: {e}")
    
    def renew_circuit(self):
        """Get a new Tor circuit (new IP). Requires stem."""
        if not STEM_AVAILABLE or Signal is None:
            logger.warning("stem not installed; cannot renew circuit")
            return False
        if not self.controller:
            logger.error("Not connected to Tor control port")
            return False
        
        try:
            self.controller.signal(Signal.NEWNYM)
            logger.info("🔄 New Tor circuit established")
            time.sleep(5)  # Wait for new circuit
            return True
        except Exception as e:
            logger.error(f"Failed to renew circuit: {e}")
            return False
    
    def get_current_ip(self) -> Optional[str]:
        """Get current Tor exit node IP"""
        try:
            # Use a service that returns your IP
            connector = aiohttp.TCPConnector()
            loop = asyncio.get_event_loop()
            
            async def fetch_ip():
                async with aiohttp.ClientSession(connector=connector) as session:
                    async with session.get('https://api.ipify.org?format=json',
                                          proxy=f'socks5://{self.proxy_host}:{self.proxy_port}') as resp:
                        data = await resp.json()
                        return data.get('ip')
            
            ip = loop.run_until_complete(fetch_ip())
            logger.info(f"Current Tor IP: {ip}")
            return ip
        except Exception as e:
            logger.error(f"Failed to get Tor IP: {e}")
            return None
    
    def disconnect(self):
        """Close Tor connection"""
        if self.controller:
            self.controller.close()
            logger.info("Disconnected from Tor")


class TorScraper:
    """Scrape dark web sites via Tor"""
    
    # Common user agents for anonymity
    USER_AGENTS = [
        'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0',
        'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:91.0) Gecko/20100101 Firefox/91.0'
    ]
    
    def __init__(self, tor_connection: TorConnection):
        self.tor = tor_connection
        self.session = None
        self.request_count = 0
        self.circuit_rotation_threshold = 10  # Rotate circuit every 10 requests
    
    async def create_session(self) -> aiohttp.ClientSession:
        """Create aiohttp session with Tor proxy"""
        connector = aiohttp.TCPConnector()
        
        self.session = aiohttp.ClientSession(
            connector=connector,
            headers={'User-Agent': random.choice(self.USER_AGENTS)}
        )
        return self.session
    
    async def fetch(self, url: str, timeout: int = 30) -> Optional[str]:
        """
        Fetch a dark web page via Tor
        
        Args:
            url: .onion URL to fetch
            timeout: Request timeout in seconds
        
        Returns:
            HTML content or None if failed
        """
        if not self.session:
            await self.create_session()
        
        # Rotate circuit periodically
        self.request_count += 1
        if self.request_count % self.circuit_rotation_threshold == 0:
            logger.info(f"Rotating Tor circuit after {self.request_count} requests")
            self.tor.renew_circuit()
        
        try:
            proxy_url = f'socks5://{self.tor.proxy_host}:{self.tor.proxy_port}'
            
            async with self.session.get(
                url,
                proxy=proxy_url,
                timeout=aiohttp.ClientTimeout(total=timeout),
                ssl=False  # .onion sites don't use traditional SSL
            ) as response:
                
                if response.status == 200:
                    content = await response.text()
                    logger.info(f"✓ Successfully fetched: {url[:50]}...")
                    return content
                else:
                    logger.warning(f"HTTP {response.status} for {url}")
                    return None
                    
        except asyncio.TimeoutError:
            logger.error(f"Timeout fetching {url}")
            return None
        except Exception as e:
            logger.error(f"Error fetching {url}: {e}")
            return None
    
    async def check_onion_status(self, url: str) -> bool:
        """Check if an .onion site is online"""
        content = await self.fetch(url, timeout=15)
        return content is not None
    
    async def close(self):
        """Close scraper session"""
        if self.session:
            await self.session.close()


class RansomwareLeakSiteMonitor:
    """Monitor ransomware group leak sites"""
    
    # Active ransomware groups (December 2025)
    KNOWN_GROUPS = {
        'lockbit': {
            'name': 'LockBit 4.0',
            'onion_urls': [
                # Note: Real URLs redacted for safety
                # In production, these would be actual .onion addresses
                'http://lockbitXXXXXXXXXXXXXXXX.onion'
            ],
            'victim_page_pattern': r'/victims',
            'active': True
        },
        'blackcat': {
            'name': 'BlackCat/ALPHV',
            'onion_urls': ['http://alphvXXXXXXXXXXXXXXXX.onion'],
            'victim_page_pattern': r'/posts',
            'active': True
        },
        'qilin': {
            'name': 'Qilin',
            'onion_urls': ['http://qilinXXXXXXXXXXXXXXXX.onion'],
            'victim_page_pattern': r'/blog',
            'active': True
        },
        'akira': {
            'name': 'Akira',
            'onion_urls': ['http://akiraXXXXXXXXXXXXXXXX.onion'],
            'victim_page_pattern': r'/victims',
            'active': True
        }
    }
    
    def __init__(self, tor_scraper: TorScraper):
        self.scraper = tor_scraper
        self.victims_database = []
        logger.info(f"Initialized ransomware monitor for {len(self.KNOWN_GROUPS)} groups")
    
    async def scan_all_groups(self) -> List[Dict]:
        """Scan all known ransomware groups"""
        all_victims = []
        
        for group_id, group_info in self.KNOWN_GROUPS.items():
            if not group_info['active']:
                continue
            
            logger.info(f"Scanning {group_info['name']}...")
            victims = await self.scan_group(group_id, group_info)
            all_victims.extend(victims)
        
        return all_victims
    
    async def scan_group(self, group_id: str, group_info: Dict) -> List[Dict]:
        """Scan a specific ransomware group's leak site"""
        victims = []
        
        for onion_url in group_info['onion_urls']:
            # Check if site is online
            is_online = await self.scraper.check_onion_status(onion_url)
            
            if not is_online:
                logger.warning(f"{group_info['name']} site appears offline")
                continue
            
            # Fetch victim listing page
            victim_page = urljoin(onion_url, group_info['victim_page_pattern'])
            html = await self.scraper.fetch(victim_page)
            
            if html:
                parsed_victims = self._parse_victim_page(html, group_id, group_info['name'])
                victims.extend(parsed_victims)
                logger.info(f"Found {len(parsed_victims)} victims for {group_info['name']}")
        
        return victims
    
    def _parse_victim_page(self, html: str, group_id: str, group_name: str) -> List[Dict]:
        """Parse victim listings from HTML"""
        victims = []
        
        # Generic parsing (each group has different HTML structure)
        # This is a simplified example - production needs group-specific parsers
        
        # Look for company names (common patterns)
        company_pattern = r'<h2[^>]*>(.*?)</h2>|<h3[^>]*>(.*?)</h3>'
        matches = re.findall(company_pattern, html, re.IGNORECASE)
        
        for match in matches:
            company_name = match[0] or match[1]
            company_name = re.sub(r'<[^>]+>', '', company_name).strip()
            
            if company_name and len(company_name) > 3:
                victim = {
                    'company': company_name,
                    'ransomware_group': group_name,
                    'group_id': group_id,
                    'discovered': time.strftime('%Y-%m-%d %H:%M:%S'),
                    'status': 'active'
                }
                victims.append(victim)
        
        return victims
    
    def check_if_monitored_org(self, company_name: str, monitored_orgs: List[str]) -> bool:
        """Check if victim is a monitored organization"""
        company_lower = company_name.lower()
        for org in monitored_orgs:
            if org.lower() in company_lower or company_lower in org.lower():
                return True
        return False


class InitialAccessBrokerMonitor:
    """Monitor Initial Access Broker (IAB) marketplaces"""
    
    # Known IAB forums/markets (URLs redacted for safety)
    IAB_MARKETS = {
        'xss_forum': {
            'name': 'XSS.is',
            'onion_url': 'http://xssXXXXXXXXXXXXXXXX.onion',
            'category': 'forum',
            'access_section': '/forum/3-access/'
        },
        'exploit_in': {
            'name': 'Exploit.in',
            'onion_url': 'http://exploitXXXXXXXXXXXXXXXX.onion',
            'category': 'market',
            'access_section': '/category/access'
        }
    }
    
    def __init__(self, tor_scraper: TorScraper):
        self.scraper = tor_scraper
        self.listings_database = []
        logger.info(f"Initialized IAB monitor for {len(self.IAB_MARKETS)} markets")
    
    async def scan_all_markets(self) -> List[DarkWebListing]:
        """Scan all IAB marketplaces"""
        all_listings = []
        
        for market_id, market_info in self.IAB_MARKETS.items():
            logger.info(f"Scanning {market_info['name']}...")
            listings = await self.scan_market(market_id, market_info)
            all_listings.extend(listings)
        
        return all_listings
    
    async def scan_market(self, market_id: str, market_info: Dict) -> List[DarkWebListing]:
        """Scan specific IAB marketplace"""
        listings = []
        
        # Check if market is online
        base_url = market_info['onion_url']
        is_online = await self.scraper.check_onion_status(base_url)
        
        if not is_online:
            logger.warning(f"{market_info['name']} appears offline")
            return listings
        
        # Fetch access listings section
        access_url = urljoin(base_url, market_info['access_section'])
        html = await self.scraper.fetch(access_url)
        
        if html:
            parsed_listings = self._parse_access_listings(html, market_id, market_info['name'])
            listings.extend(parsed_listings)
            logger.info(f"Found {len(parsed_listings)} IAB listings on {market_info['name']}")
        
        return listings
    
    def _parse_access_listings(self, html: str, market_id: str, market_name: str) -> List[DarkWebListing]:
        """Parse IAB access listings"""
        listings = []
        
        # Generic parsing - production needs market-specific parsers
        # Look for common patterns in IAB listings
        
        # Pattern: [COMPANY] RDP/VPN Access - $XXXX
        listing_pattern = r'(?:RDP|VPN|Domain|Admin).*?(\$\d+(?:,\d+)?(?:\.\d+)?)'
        matches = re.finditer(listing_pattern, html, re.IGNORECASE)
        
        for i, match in enumerate(matches):
            # Extract surrounding text for context
            start = max(0, match.start() - 100)
            end = min(len(html), match.end() + 100)
            context = html[start:end]
            
            # Clean HTML tags
            context_clean = re.sub(r'<[^>]+>', ' ', context).strip()
            
            # Try to extract price
            price_match = re.search(r'\$(\d+(?:,\d+)?(?:\.\d+)?)', match.group())
            price = float(price_match.group(1).replace(',', '')) if price_match else None
            
            listing = DarkWebListing(
                title=f"Access Listing #{i+1}",
                description=context_clean[:200],
                price=price,
                seller="Unknown",  # Would parse from HTML
                category="initial_access",
                source_url=f"{market_name}",
                posted_date=None,
                data_type="network_access",
                raw_html=context
            )
            listings.append(listing)
        
        return listings
    
    def calculate_risk_score(self, listing: DarkWebListing) -> int:
        """Calculate risk score for an IAB listing"""
        score = 0
        
        # Price-based risk
        if listing.price:
            if listing.price > 100000:
                score += 50  # Enterprise, likely domain admin
            elif listing.price > 50000:
                score += 30
            elif listing.price > 10000:
                score += 15
        
        # Access type risk
        desc_lower = listing.description.lower()
        if 'domain admin' in desc_lower:
            score += 40
        elif 'admin' in desc_lower:
            score += 25
        elif 'vpn' in desc_lower or 'rdp' in desc_lower:
            score += 15
        
        return min(score, 100)


class CredentialMarketMonitor:
    """Monitor dark web credential marketplaces"""
    
    def __init__(self, tor_scraper: TorScraper):
        self.scraper = tor_scraper
        logger.info("Initialized credential market monitor")
    
    async def search_for_domain(self, domain: str, markets: List[str] = None) -> List[Dict]:
        """Search credential markets for specific domain"""
        findings = []
        
        # Placeholder - would implement actual market searches
        logger.info(f"Searching credential markets for: {domain}")
        
        # In production: search Telegram channels, paste sites, stealer log shops
        # that are accessible via Tor
        
        return findings


class DarkWebIntelligencePlatform:
    """Main dark web intelligence orchestrator"""
    
    def __init__(self, 
                 monitored_domains: List[str],
                 monitored_organizations: List[str],
                 tor_config: Dict = None):
        """
        Initialize dark web intelligence platform
        
        Args:
            monitored_domains: Domains to monitor for leaks
            monitored_organizations: Company names to watch
            tor_config: Tor connection config
        """
        tor_config = tor_config or {}
        
        # Initialize Tor
        self.tor = TorConnection(
            tor_proxy_host=tor_config.get('proxy_host', '127.0.0.1'),
            tor_proxy_port=tor_config.get('proxy_port', 9050),
            tor_control_port=tor_config.get('control_port', 9051),
            tor_password=tor_config.get('password')
        )
        
        self.scraper = TorScraper(self.tor)
        
        # Initialize monitors
        self.ransomware_monitor = RansomwareLeakSiteMonitor(self.scraper)
        self.iab_monitor = InitialAccessBrokerMonitor(self.scraper)
        self.credential_monitor = CredentialMarketMonitor(self.scraper)
        
        # Configuration
        self.monitored_domains = [d.lower() for d in monitored_domains]
        self.monitored_organizations = [o.lower() for o in monitored_organizations]
        
        # Results storage
        self.critical_findings = []
        
        logger.info("Dark Web Intelligence Platform initialized")
    
    async def initialize(self) -> bool:
        """Initialize Tor connection"""
        logger.info("Connecting to Tor network...")
        success = await self.tor.connect()
        
        if success:
            # Test connection
            current_ip = self.tor.get_current_ip()
            if current_ip:
                logger.info(f"✓ Tor connection verified (Exit IP: {current_ip})")
        
        return success
    
    async def run_full_scan(self) -> Dict:
        """Run comprehensive dark web scan"""
        logger.info("="*80)
        logger.info("🔍 Starting Full Dark Web Intelligence Scan")
        logger.info("="*80)
        
        results = {
            'ransomware_victims': [],
            'iab_listings': [],
            'credential_leaks': [],
            'critical_alerts': []
        }
        
        # 1. Scan ransomware leak sites
        logger.info("\n📊 Phase 1: Ransomware Leak Site Monitoring")
        logger.info("-" * 40)
        victims = await self.ransomware_monitor.scan_all_groups()
        results['ransomware_victims'] = victims
        
        # Check for monitored organizations
        for victim in victims:
            if self.ransomware_monitor.check_if_monitored_org(
                victim['company'], 
                self.monitored_organizations
            ):
                alert = {
                    'severity': 'CRITICAL',
                    'type': 'ransomware_victim',
                    'title': f"🚨 Monitored Organization on Ransomware Leak Site",
                    'details': victim
                }
                results['critical_alerts'].append(alert)
                logger.error(f"🚨 CRITICAL: {victim['company']} found on {victim['ransomware_group']} leak site!")
        
        # 2. Scan IAB marketplaces
        logger.info("\n📊 Phase 2: Initial Access Broker Monitoring")
        logger.info("-" * 40)
        iab_listings = await self.iab_monitor.scan_all_markets()
        results['iab_listings'] = iab_listings
        
        # Analyze high-risk listings
        for listing in iab_listings:
            risk_score = self.iab_monitor.calculate_risk_score(listing)
            if risk_score >= 50:
                results['critical_alerts'].append({
                    'severity': 'HIGH',
                    'type': 'high_value_access',
                    'title': f"High-Value Access Listing (Risk: {risk_score})",
                    'details': listing.__dict__
                })
        
        # 3. Search credential markets
        logger.info("\n📊 Phase 3: Credential Market Monitoring")
        logger.info("-" * 40)
        for domain in self.monitored_domains:
            creds = await self.credential_monitor.search_for_domain(domain)
            if creds:
                results['credential_leaks'].extend(creds)
        
        # Summary
        logger.info("\n" + "="*80)
        logger.info("📈 SCAN SUMMARY")
        logger.info("="*80)
        logger.info(f"Ransomware Victims Found: {len(results['ransomware_victims'])}")
        logger.info(f"IAB Listings Found: {len(results['iab_listings'])}")
        logger.info(f"Credential Leaks Found: {len(results['credential_leaks'])}")
        logger.info(f"🚨 CRITICAL ALERTS: {len(results['critical_alerts'])}")
        logger.info("="*80)
        
        return results
    
    async def continuous_monitoring(self, interval_hours: int = 6):
        """Run continuous dark web monitoring"""
        logger.info(f"Starting continuous monitoring (interval: {interval_hours} hours)")
        
        while True:
            try:
                await self.run_full_scan()
                logger.info(f"\nNext scan in {interval_hours} hours...")
                await asyncio.sleep(interval_hours * 3600)
                
                # Rotate Tor circuit between scans
                self.tor.renew_circuit()
                
            except KeyboardInterrupt:
                logger.info("Monitoring stopped by user")
                break
            except Exception as e:
                logger.error(f"Error in monitoring loop: {e}")
                await asyncio.sleep(1800)  # Wait 30 min on error
    
    async def shutdown(self):
        """Cleanup and shutdown"""
        await self.scraper.close()
        self.tor.disconnect()
        logger.info("Platform shutdown complete")


# Demo and testing functions
async def demo_tor_connection():
    """Demo: Test Tor connection"""
    print("\n🌐 Testing Tor Connection")
    print("="*80)
    
    tor = TorConnection()
    
    if await tor.connect():
        print("✓ Successfully connected to Tor network")
        
        # Get current IP
        current_ip = tor.get_current_ip()
        print(f"✓ Current exit node IP: {current_ip}")
        
        # Test circuit rotation
        print("\n🔄 Testing circuit rotation...")
        tor.renew_circuit()
        new_ip = tor.get_current_ip()
        print(f"✓ New exit node IP: {new_ip}")
        
        if current_ip != new_ip:
            print("✓ Circuit rotation successful!")
        
        tor.disconnect()
    else:
        print("❌ Failed to connect to Tor")
        print("\nTroubleshooting:")
        print("1. Install Tor: brew install tor (Mac) or apt install tor (Linux)")
        print("2. Start Tor: tor or brew services start tor")
        print("3. Verify Tor is running: lsof -i :9050")


async def demo_dark_web_scan():
    """Demo: Run a dark web intelligence scan"""
    print("\n🕵️ Dark Web Intelligence Scan Demo")
    print("="*80)
    
    # Configuration
    monitored_domains = ['example.com', 'company.com']
    monitored_orgs = ['Acme Corporation', 'Example Industries']
    
    # Initialize platform
    platform = DarkWebIntelligencePlatform(
        monitored_domains=monitored_domains,
        monitored_organizations=monitored_orgs
    )
    
    # Connect to Tor
    if not await platform.initialize():
        print("❌ Failed to initialize Tor connection")
        return
    
    # Run scan
    try:
        results = await platform.run_full_scan()
        
        # Display critical alerts
        if results['critical_alerts']:
            print("\n🚨 CRITICAL ALERTS:")
            for alert in results['critical_alerts']:
                print(f"\n{alert['severity']}: {alert['title']}")
                print(f"Type: {alert['type']}")
        
    finally:
        await platform.shutdown()


async def demo_ransomware_monitoring():
    """Demo: Ransomware leak site monitoring"""
    print("\n🦠 Ransomware Leak Site Monitoring Demo")
    print("="*80)
    
    tor = TorConnection()
    
    if not await tor.connect():
        print("❌ Failed to connect to Tor")
        return
    
    scraper = TorScraper(tor)
    monitor = RansomwareLeakSiteMonitor(scraper)
    
    try:
        print("\nScanning known ransomware groups...")
        victims = await monitor.scan_all_groups()
        
        print(f"\nTotal victims found: {len(victims)}")
        
        # Display sample
        if victims:
            print("\nSample victims:")
            for victim in victims[:5]:
                print(f"  - {victim['company']} ({victim['ransomware_group']})")
        
    finally:
        await scraper.close()
        tor.disconnect()


def print_setup_instructions():
    """Print Tor setup instructions"""
    print("\n" + "="*80)
    print("🔧 TOR SETUP INSTRUCTIONS")
    print("="*80)
    print("\n1. Install Tor:")
    print("   macOS:   brew install tor")
    print("   Linux:   sudo apt install tor")
    print("   Windows: Download from https://www.torproject.org/")
    
    print("\n2. Start Tor:")
    print("   macOS:   brew services start tor")
    print("   Linux:   sudo systemctl start tor")
    print("   Windows: Run Tor Browser or tor.exe")
    
    print("\n3. Verify Tor is running:")
    print("   Command: lsof -i :9050  (should show tor process)")
    
    print("\n4. Configure Tor control port (optional):")
    print("   Edit /usr/local/etc/tor/torrc (macOS)")
    print("   or /etc/tor/torrc (Linux)")
    print("   Add: ControlPort 9051")
    print("   Add: HashedControlPassword (generate with: tor --hash-password yourpassword)")
    
    print("\n5. Install Python dependencies:")
    print("   pip install aiohttp stem PySocks")
    
    print("\n" + "="*80)


if __name__ == "__main__":
    print("ShadowHunter - Tor Network Integration Module")
    print("\nSelect demo mode:")
    print("1. Setup Instructions")
    print("2. Test Tor Connection")
    print("3. Ransomware Monitoring Demo")
    print("4. Full Dark Web Scan (requires Tor running)")
    
    choice = input("\nEnter choice (1-4): ").strip()
    
    if choice == "1":
        print_setup_instructions()
    elif choice == "2":
        asyncio.run(demo_tor_connection())
    elif choice == "3":
        asyncio.run(demo_ransomware_monitoring())
    elif choice == "4":
        asyncio.run(demo_dark_web_scan())
    else:
        print("Invalid choice")
        print_setup_instructions()