shadowhunter_core.py

#!/usr/bin/env python3
"""
ShadowHunter - Dark Web Credential Intelligence Platform
Module: Core Credential Monitor
Author: Fevra
Version: 0.1.0
"""

import asyncio
import aiohttp
import hashlib
import json
import re
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Set
from dataclasses import dataclass, asdict
from enum import Enum
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('ShadowHunter')


class ThreatSeverity(Enum):
    """Threat severity levels"""
    CRITICAL = "CRITICAL"
    HIGH = "HIGH"
    MEDIUM = "MEDIUM"
    LOW = "LOW"
    INFO = "INFO"


class SourceType(Enum):
    """Data source types"""
    PASTEBIN = "pastebin"
    TELEGRAM = "telegram"
    BREACH_DB = "breach_database"
    STEALER_LOG = "stealer_log"
    DARKWEB_FORUM = "darkweb_forum"
    HAVEIBEENPWNED = "haveibeenpwned"


@dataclass
class Credential:
    """Credential data structure"""
    email: str
    password: Optional[str] = None
    domain: Optional[str] = None
    source: str = "unknown"
    source_type: SourceType = SourceType.BREACH_DB
    timestamp: str = ""
    additional_data: Dict = None
    
    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.utcnow().isoformat()
        if not self.domain and self.email:
            self.domain = self.email.split('@')[1] if '@' in self.email else None
        if self.additional_data is None:
            self.additional_data = {}
    
    def hash_email(self) -> str:
        """Create searchable hash of email"""
        return hashlib.sha256(self.email.encode()).hexdigest()
    
    def to_dict(self) -> Dict:
        """Convert to dictionary"""
        data = asdict(self)
        data['source_type'] = self.source_type.value
        data['email_hash'] = self.hash_email()
        return data


@dataclass
class ThreatAlert:
    """Threat alert structure"""
    severity: ThreatSeverity
    title: str
    description: str
    credentials: List[Credential]
    affected_domain: str
    timestamp: str = ""
    indicators: Dict = None
    
    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.utcnow().isoformat()
        if self.indicators is None:
            self.indicators = {}
    
    def to_dict(self) -> Dict:
        """Convert to dictionary"""
        return {
            'severity': self.severity.value,
            'title': self.title,
            'description': self.description,
            'credentials_count': len(self.credentials),
            'affected_domain': self.affected_domain,
            'timestamp': self.timestamp,
            'indicators': self.indicators
        }


class CredentialExtractor:
    """Extract credentials from various text formats"""
    
    # Regex patterns
    EMAIL_PATTERN = re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b')
    COMBO_PATTERN = re.compile(r'([^\s:]+@[^\s:]+):([^\s]+)')
    
    @staticmethod
    def extract_emails(text: str) -> List[str]:
        """Extract email addresses from text"""
        return list(set(CredentialExtractor.EMAIL_PATTERN.findall(text)))
    
    @staticmethod
    def extract_combos(text: str) -> List[Credential]:
        """Extract email:password combinations"""
        combos = CredentialExtractor.COMBO_PATTERN.findall(text)
        credentials = []
        
        for email, password in combos:
            # Basic validation
            if '@' in email and len(password) > 3:
                cred = Credential(
                    email=email.strip(),
                    password=password.strip(),
                    source="combo_list"
                )
                credentials.append(cred)
        
        return credentials
    
    @staticmethod
    def parse_breach_format(text: str) -> List[Credential]:
        """Parse common breach database formats"""
        credentials = []
        lines = text.split('\n')
        
        for line in lines:
            line = line.strip()
            if not line:
                continue
            
            # Try different formats
            if ':' in line:
                parts = line.split(':', 1)
                if len(parts) == 2 and '@' in parts[0]:
                    cred = Credential(
                        email=parts[0].strip(),
                        password=parts[1].strip(),
                        source="breach_db"
                    )
                    credentials.append(cred)
            elif '@' in line:
                # Email only
                cred = Credential(
                    email=line.strip(),
                    source="breach_db"
                )
                credentials.append(cred)
        
        return credentials


class DomainMonitor:
    """Monitor specific domains for credential leaks"""
    
    def __init__(self, watched_domains: List[str]):
        self.watched_domains = set(d.lower() for d in watched_domains)
        self.findings: List[Credential] = []
        logger.info(f"Initialized DomainMonitor with {len(self.watched_domains)} domains")
    
    def is_monitored(self, email: str) -> bool:
        """Check if email belongs to monitored domain"""
        if '@' not in email:
            return False
        domain = email.split('@')[1].lower()
        return domain in self.watched_domains
    
    def add_finding(self, credential: Credential):
        """Add a credential finding"""
        if self.is_monitored(credential.email):
            self.findings.append(credential)
            logger.warning(f"🚨 MONITORED DOMAIN LEAK: {credential.email}")
            return True
        return False
    
    def get_findings_by_domain(self, domain: str) -> List[Credential]:
        """Get all findings for a specific domain"""
        return [c for c in self.findings if c.domain == domain.lower()]
    
    def get_summary(self) -> Dict:
        """Get summary of findings"""
        summary = {}
        for domain in self.watched_domains:
            findings = self.get_findings_by_domain(domain)
            summary[domain] = {
                'count': len(findings),
                'credentials': [c.to_dict() for c in findings[:5]]  # First 5
            }
        return summary


class PastebinMonitor:
    """Monitor Pastebin for credential leaks"""
    
    BASE_URL = "https://pastebin.com"
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key
        self.seen_pastes: Set[str] = set()
        logger.info("Initialized PastebinMonitor")
    
    async def fetch_recent_pastes(self, session: aiohttp.ClientSession) -> List[Dict]:
        """Fetch recent public pastes"""
        # Note: Real implementation requires Pastebin API key
        # This is a placeholder showing the structure
        
        # Simulated paste data for demonstration
        demo_pastes = [
            {
                'key': 'demo123',
                'title': 'Database Dump 2025',
                'size': '15000',
                'expire_date': '0',
                'url': f'{self.BASE_URL}/demo123'
            }
        ]
        
        logger.info(f"Fetched {len(demo_pastes)} recent pastes")
        return demo_pastes
    
    async def analyze_paste(self, paste_data: Dict, session: aiohttp.ClientSession) -> List[Credential]:
        """Analyze a paste for credentials"""
        paste_key = paste_data['key']
        
        if paste_key in self.seen_pastes:
            return []
        
        self.seen_pastes.add(paste_key)
        
        # Simulated paste content for demonstration
        # In production: fetch actual paste content
        demo_content = """
        admin@company.com:Password123
        user@example.com:SecurePass456
        test@corporation.com:TestPass789
        """
        
        # Extract credentials
        credentials = CredentialExtractor.extract_combos(demo_content)
        
        for cred in credentials:
            cred.source = f"pastebin:{paste_key}"
            cred.source_type = SourceType.PASTEBIN
            cred.additional_data = {
                'paste_title': paste_data.get('title', 'Untitled'),
                'paste_url': paste_data['url']
            }
        
        if credentials:
            logger.info(f"Found {len(credentials)} credentials in paste {paste_key}")
        
        return credentials


class HaveIBeenPwnedMonitor:
    """Monitor HaveIBeenPwned API for breaches"""
    
    BASE_URL = "https://haveibeenpwned.com/api/v3"
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key
        self.headers = {}
        if api_key:
            self.headers['hibp-api-key'] = api_key
        logger.info("Initialized HaveIBeenPwnedMonitor")
    
    async def check_email(self, email: str, session: aiohttp.ClientSession) -> Dict:
        """Check if email appears in breaches"""
        if not self.api_key:
            logger.warning("HIBP API key not configured")
            return {'breaches': [], 'pastes': []}
        
        url = f"{self.BASE_URL}/breachedaccount/{email}"
        
        try:
            async with session.get(url, headers=self.headers) as response:
                if response.status == 200:
                    breaches = await response.json()
                    logger.info(f"Email {email} found in {len(breaches)} breaches")
                    return {'breaches': breaches, 'pastes': []}
                elif response.status == 404:
                    return {'breaches': [], 'pastes': []}
                else:
                    logger.error(f"HIBP API error: {response.status}")
                    return {'breaches': [], 'pastes': []}
        except Exception as e:
            logger.error(f"Error checking HIBP: {e}")
            return {'breaches': [], 'pastes': []}
    
    async def check_domain(self, domain: str, session: aiohttp.ClientSession) -> List[Credential]:
        """Check for domain breaches"""
        # Note: Domain search requires special HIBP subscription
        # This is a placeholder showing the structure
        
        logger.info(f"Checking domain: {domain}")
        return []


class ThreatCorrelator:
    """Correlate findings and generate alerts"""
    
    def __init__(self, domain_monitor: DomainMonitor):
        self.domain_monitor = domain_monitor
        self.alert_threshold = {
            ThreatSeverity.CRITICAL: 10,
            ThreatSeverity.HIGH: 5,
            ThreatSeverity.MEDIUM: 2
        }
    
    def analyze_findings(self, credentials: List[Credential]) -> List[ThreatAlert]:
        """Analyze credential findings and generate alerts"""
        alerts = []
        
        # Group by domain
        domain_groups = {}
        for cred in credentials:
            if cred.domain:
                if cred.domain not in domain_groups:
                    domain_groups[cred.domain] = []
                domain_groups[cred.domain].append(cred)
        
        # Generate alerts for monitored domains
        for domain, domain_creds in domain_groups.items():
            if not self.domain_monitor.is_monitored(domain_creds[0].email):
                continue
            
            count = len(domain_creds)
            
            # Determine severity based on count
            if count >= self.alert_threshold[ThreatSeverity.CRITICAL]:
                severity = ThreatSeverity.CRITICAL
            elif count >= self.alert_threshold[ThreatSeverity.HIGH]:
                severity = ThreatSeverity.HIGH
            elif count >= self.alert_threshold[ThreatSeverity.MEDIUM]:
                severity = ThreatSeverity.MEDIUM
            else:
                severity = ThreatSeverity.LOW
            
            # Check for additional risk factors
            indicators = self._calculate_risk_indicators(domain_creds)
            
            alert = ThreatAlert(
                severity=severity,
                title=f"Credential Leak Detected: {domain}",
                description=f"Found {count} leaked credentials for monitored domain {domain}",
                credentials=domain_creds,
                affected_domain=domain,
                indicators=indicators
            )
            
            alerts.append(alert)
            logger.warning(f"Generated {severity.value} alert for {domain}")
        
        return alerts
    
    def _calculate_risk_indicators(self, credentials: List[Credential]) -> Dict:
        """Calculate risk indicators for credentials"""
        indicators = {
            'has_passwords': sum(1 for c in credentials if c.password) > 0,
            'unique_sources': len(set(c.source for c in credentials)),
            'recent_leak': any(self._is_recent(c) for c in credentials),
            'executive_accounts': self._check_executive_accounts(credentials)
        }
        
        return indicators
    
    def _is_recent(self, credential: Credential) -> bool:
        """Check if credential leak is recent (last 7 days)"""
        try:
            cred_time = datetime.fromisoformat(credential.timestamp)
            return datetime.utcnow() - cred_time < timedelta(days=7)
        except:
            return False
    
    def _check_executive_accounts(self, credentials: List[Credential]) -> bool:
        """Check for executive-level accounts"""
        executive_keywords = ['ceo', 'cto', 'cfo', 'ciso', 'admin', 'director']
        for cred in credentials:
            email_lower = cred.email.lower()
            if any(keyword in email_lower for keyword in executive_keywords):
                return True
        return False


class AlertManager:
    """Manage and display threat alerts"""
    
    def __init__(self):
        self.alerts: List[ThreatAlert] = []
    
    def add_alert(self, alert: ThreatAlert):
        """Add a new alert"""
        self.alerts.append(alert)
        self._display_alert(alert)
    
    def _display_alert(self, alert: ThreatAlert):
        """Display alert to console"""
        severity_emoji = {
            ThreatSeverity.CRITICAL: "🔴",
            ThreatSeverity.HIGH: "🟠",
            ThreatSeverity.MEDIUM: "🟡",
            ThreatSeverity.LOW: "🟢",
            ThreatSeverity.INFO: "ℹ️"
        }
        
        print("\n" + "="*80)
        print(f"{severity_emoji[alert.severity]} {alert.severity.value} THREAT ALERT")
        print("="*80)
        print(f"Title: {alert.title}")
        print(f"Domain: {alert.affected_domain}")
        print(f"Credentials Found: {len(alert.credentials)}")
        print(f"Timestamp: {alert.timestamp}")
        print("\nRisk Indicators:")
        for indicator, value in alert.indicators.items():
            print(f"  - {indicator}: {value}")
        print("\nSample Credentials:")
        for i, cred in enumerate(alert.credentials[:3], 1):
            print(f"  {i}. {cred.email} (Source: {cred.source})")
        if len(alert.credentials) > 3:
            print(f"  ... and {len(alert.credentials) - 3} more")
        print("="*80 + "\n")
    
    def get_summary(self) -> Dict:
        """Get alert summary"""
        return {
            'total_alerts': len(self.alerts),
            'by_severity': {
                severity.value: sum(1 for a in self.alerts if a.severity == severity)
                for severity in ThreatSeverity
            },
            'recent_alerts': [a.to_dict() for a in self.alerts[-5:]]
        }


class ShadowHunter:
    """Main ShadowHunter orchestrator"""
    
    def __init__(self, watched_domains: List[str], api_keys: Dict[str, str] = None):
        """
        Initialize ShadowHunter
        
        Args:
            watched_domains: List of domains to monitor
            api_keys: Dict of API keys {'pastebin': 'key', 'hibp': 'key'}
        """
        api_keys = api_keys or {}
        
        # Core components
        self.domain_monitor = DomainMonitor(watched_domains)
        self.pastebin_monitor = PastebinMonitor(api_keys.get('pastebin'))
        self.hibp_monitor = HaveIBeenPwnedMonitor(api_keys.get('hibp'))
        self.correlator = ThreatCorrelator(self.domain_monitor)
        self.alert_manager = AlertManager()
        
        logger.info("ShadowHunter initialized successfully")
    
    async def scan_all_sources(self):
        """Scan all configured sources"""
        all_credentials = []
        
        async with aiohttp.ClientSession() as session:
            # Scan Pastebin
            logger.info("Scanning Pastebin...")
            pastes = await self.pastebin_monitor.fetch_recent_pastes(session)
            for paste in pastes:
                creds = await self.pastebin_monitor.analyze_paste(paste, session)
                all_credentials.extend(creds)
            
            # Check HIBP for monitored domains
            logger.info("Checking HaveIBeenPwned...")
            for domain in self.domain_monitor.watched_domains:
                creds = await self.hibp_monitor.check_domain(domain, session)
                all_credentials.extend(creds)
        
        return all_credentials
    
    async def run_scan(self):
        """Run a single scan cycle"""
        logger.info("Starting scan cycle...")
        
        # Collect credentials from all sources
        credentials = await self.scan_all_sources()
        
        # Filter for monitored domains
        monitored_creds = [c for c in credentials if self.domain_monitor.add_finding(c)]
        
        logger.info(f"Found {len(credentials)} total credentials, {len(monitored_creds)} for monitored domains")
        
        # Generate alerts
        if monitored_creds:
            alerts = self.correlator.analyze_findings(monitored_creds)
            for alert in alerts:
                self.alert_manager.add_alert(alert)
        
        return {
            'credentials_found': len(credentials),
            'monitored_credentials': len(monitored_creds),
            'alerts_generated': len(self.alert_manager.alerts)
        }
    
    async def run_continuous(self, interval_minutes: int = 60):
        """Run continuous monitoring"""
        logger.info(f"Starting continuous monitoring (interval: {interval_minutes} minutes)")
        
        while True:
            try:
                await self.run_scan()
                logger.info(f"Scan complete. Next scan in {interval_minutes} minutes.")
                await asyncio.sleep(interval_minutes * 60)
            except KeyboardInterrupt:
                logger.info("Monitoring stopped by user")
                break
            except Exception as e:
                logger.error(f"Error in monitoring loop: {e}")
                await asyncio.sleep(300)  # Wait 5 minutes before retry
    
    def get_status(self) -> Dict:
        """Get current status"""
        return {
            'monitored_domains': list(self.domain_monitor.watched_domains),
            'total_findings': len(self.domain_monitor.findings),
            'alerts': self.alert_manager.get_summary(),
            'domain_summary': self.domain_monitor.get_summary()
        }


# Demo/Testing Functions
async def demo_basic_usage():
    """Demonstrate basic usage"""
    print("\n🔥 ShadowHunter - Dark Web Credential Intelligence Platform")
    print("="*80)
    
    # Initialize with monitored domains
    watched_domains = [
        'company.com',
        'example.com',
        'corporation.com'
    ]
    
    hunter = ShadowHunter(
        watched_domains=watched_domains,
        api_keys={
            'pastebin': 'YOUR_API_KEY_HERE',  # Optional
            'hibp': 'YOUR_API_KEY_HERE'       # Optional
        }
    )
    
    # Run a single scan
    print("\n📡 Running scan...")
    results = await hunter.run_scan()
    
    print(f"\n📊 Scan Results:")
    print(f"  - Total credentials found: {results['credentials_found']}")
    print(f"  - Monitored credentials: {results['monitored_credentials']}")
    print(f"  - Alerts generated: {results['alerts_generated']}")
    
    # Display status
    status = hunter.get_status()
    print(f"\n📈 Current Status:")
    print(json.dumps(status, indent=2))


def demo_credential_extraction():
    """Demonstrate credential extraction"""
    print("\n🔍 Credential Extraction Demo")
    print("="*80)
    
    # Sample breach data
    sample_breach = """
    Database Dump - Company XYZ
    admin@company.com:AdminPass123
    user1@example.com:Password456
    ceo@corporation.com:SecurePass789
    test@company.com:TestPass000
    """
    
    # Extract credentials
    credentials = CredentialExtractor.extract_combos(sample_breach)
    
    print(f"\nExtracted {len(credentials)} credentials:")
    for i, cred in enumerate(credentials, 1):
        print(f"{i}. {cred.email} | Domain: {cred.domain} | Has Password: {bool(cred.password)}")


if __name__ == "__main__":
    print("ShadowHunter - Credential Monitor Module")
    print("Choose demo mode:")
    print("1. Credential Extraction Demo (no API required)")
    print("2. Full Scan Demo (simulated)")
    
    choice = input("\nEnter choice (1-2): ").strip()
    
    if choice == "1":
        demo_credential_extraction()
    elif choice == "2":
        asyncio.run(demo_basic_usage())
    else:
        print("Invalid choice. Running credential extraction demo...")
        demo_credential_extraction()