shadowhunter_osint.py

#!/usr/bin/env python3
"""
ShadowHunter - Dark Web Credential Intelligence Platform
Module: OSINT Lookup Engine (Email, Phone, Username, Domain)
Author: Fevra
Version: 0.1.0

Comprehensive OSINT lookup capabilities:
- Email intelligence (breach check, domain info, disposable detection)
- Phone number analysis (carrier, location, type)
- Username enumeration across platforms
- Domain reconnaissance (WHOIS, DNS, subdomains)
- Social media profile discovery
- Reverse image search integration
"""

import asyncio
import aiohttp
import re
import socket
import hashlib
import json
from datetime import datetime, timezone
from typing import List, Dict, Optional, Any, Set, Tuple
from dataclasses import dataclass, field
from urllib.parse import quote_plus, urlparse
from pathlib import Path
from enum import Enum
import dns.resolver
try:
    import whois
    WHOIS_AVAILABLE = True
except ImportError:
    whois = None
    WHOIS_AVAILABLE = False

# Import logger
try:
    from shadowhunter_logger import get_logger, LogLevel
except ImportError:
    import logging
    def get_logger(name, **kwargs):
        return logging.getLogger(name)

# Initialize module logger
logger = get_logger("OSINT", log_level=LogLevel.DEBUG)


# ============================================================================
# CONFIGURATION & CONSTANTS
# ============================================================================

class DisposableEmailDomains:
    """Known disposable/temporary email domains."""
    DOMAINS = {
        '10minutemail.com', 'guerrillamail.com', 'mailinator.com',
        'tempmail.com', 'throwaway.email', 'temp-mail.org', 'fakeinbox.com',
        'maildrop.cc', 'getairmail.com', 'mohmal.com', 'trashmail.com',
        'yopmail.com', 'sharklasers.com', 'getnada.com', 'tempinbox.com',
        'emailondeck.com', 'burnermail.io', 'tempmailaddress.com',
        'mintemail.com', 'discard.email', 'spamgourmet.com'
    }


class SocialPlatforms:
    """Social media platforms for username enumeration."""
    PLATFORMS = {
        "twitter": "https://twitter.com/{username}",
        "instagram": "https://instagram.com/{username}",
        "facebook": "https://facebook.com/{username}",
        "linkedin": "https://linkedin.com/in/{username}",
        "github": "https://github.com/{username}",
        "reddit": "https://reddit.com/user/{username}",
        "tiktok": "https://tiktok.com/@{username}",
        "youtube": "https://youtube.com/@{username}",
        "pinterest": "https://pinterest.com/{username}",
        "tumblr": "https://{username}.tumblr.com",
        "medium": "https://medium.com/@{username}",
        "devto": "https://dev.to/{username}",
        "hackernews": "https://news.ycombinator.com/user?id={username}",
        "keybase": "https://keybase.io/{username}",
        "telegram": "https://t.me/{username}",
        "snapchat": "https://snapchat.com/add/{username}",
        "twitch": "https://twitch.tv/{username}",
        "patreon": "https://patreon.com/{username}",
        "spotify": "https://open.spotify.com/user/{username}",
        "soundcloud": "https://soundcloud.com/{username}",
        "vimeo": "https://vimeo.com/{username}",
        "flickr": "https://flickr.com/people/{username}",
        "dribbble": "https://dribbble.com/{username}",
        "behance": "https://behance.net/{username}",
        "gitlab": "https://gitlab.com/{username}",
        "bitbucket": "https://bitbucket.org/{username}",
        "npm": "https://npmjs.com/~{username}",
        "pypi": "https://pypi.org/user/{username}",
    }


# ============================================================================
# DATA MODELS
# ============================================================================

class RiskLevel(str, Enum):
    """Risk level classification."""
    CRITICAL = "CRITICAL"
    HIGH = "HIGH"
    MEDIUM = "MEDIUM"
    LOW = "LOW"
    INFO = "INFO"


@dataclass
class BreachRecord:
    """Data breach record."""
    name: str
    date: Optional[str] = None
    records_count: Optional[int] = None
    data_types: List[str] = field(default_factory=list)
    description: Optional[str] = None


@dataclass
class EmailIntelligence:
    """Email address intelligence report."""
    email: str
    valid_format: bool = False
    disposable: bool = False
    domain: str = ""
    domain_exists: bool = False
    mx_records: List[str] = field(default_factory=list)
    breaches: List[BreachRecord] = field(default_factory=list)
    risk_level: RiskLevel = RiskLevel.INFO
    risk_factors: List[str] = field(default_factory=list)
    social_profiles: List[Dict[str, str]] = field(default_factory=list)
    gravatar_url: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "email": self.email,
            "valid_format": self.valid_format,
            "disposable": self.disposable,
            "domain": self.domain,
            "domain_exists": self.domain_exists,
            "mx_records": self.mx_records,
            "breaches": [
                {"name": b.name, "date": b.date, "records": b.records_count}
                for b in self.breaches
            ],
            "risk_level": self.risk_level.value,
            "risk_factors": self.risk_factors,
            "social_profiles": self.social_profiles,
            "gravatar_url": self.gravatar_url
        }


@dataclass
class PhoneIntelligence:
    """Phone number intelligence report."""
    number: str
    valid: bool = False
    country_code: Optional[str] = None
    country_name: Optional[str] = None
    carrier: Optional[str] = None
    line_type: Optional[str] = None  # mobile, landline, voip
    region: Optional[str] = None
    timezone: Optional[str] = None
    risk_level: RiskLevel = RiskLevel.INFO
    risk_factors: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "number": self.number,
            "valid": self.valid,
            "country_code": self.country_code,
            "country_name": self.country_name,
            "carrier": self.carrier,
            "line_type": self.line_type,
            "region": self.region,
            "timezone": self.timezone,
            "risk_level": self.risk_level.value,
            "risk_factors": self.risk_factors
        }


@dataclass
class UsernameIntelligence:
    """Username enumeration report."""
    username: str
    found_platforms: List[Dict[str, str]] = field(default_factory=list)
    not_found_platforms: List[str] = field(default_factory=list)
    possible_emails: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "username": self.username,
            "found_count": len(self.found_platforms),
            "platforms": self.found_platforms,
            "possible_emails": self.possible_emails
        }


@dataclass
class DomainIntelligence:
    """Domain reconnaissance report."""
    domain: str
    registered: bool = False
    registrar: Optional[str] = None
    creation_date: Optional[str] = None
    expiration_date: Optional[str] = None
    nameservers: List[str] = field(default_factory=list)
    a_records: List[str] = field(default_factory=list)
    mx_records: List[str] = field(default_factory=list)
    txt_records: List[str] = field(default_factory=list)
    subdomains: List[str] = field(default_factory=list)
    technologies: List[str] = field(default_factory=list)
    risk_level: RiskLevel = RiskLevel.INFO
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "domain": self.domain,
            "registered": self.registered,
            "registrar": self.registrar,
            "creation_date": self.creation_date,
            "expiration_date": self.expiration_date,
            "nameservers": self.nameservers,
            "a_records": self.a_records,
            "mx_records": self.mx_records,
            "txt_records": self.txt_records,
            "subdomains": self.subdomains,
            "technologies": self.technologies,
            "risk_level": self.risk_level.value
        }


# ============================================================================
# EMAIL INTELLIGENCE
# ============================================================================

class EmailLookup:
    """
    Email address intelligence gathering.
    
    Features:
    - Email format validation
    - Disposable email detection
    - Domain verification (MX records)
    - Breach database lookup
    - Gravatar lookup
    - Social profile discovery
    """
    
    # Email regex pattern
    EMAIL_PATTERN = re.compile(
        r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    )
    
    def __init__(self, hibp_api_key: Optional[str] = None):
        """
        Initialize email lookup.
        
        Args:
            hibp_api_key: HaveIBeenPwned API key for breach lookup
        """
        self.hibp_api_key = hibp_api_key
        logger.info("EmailLookup initialized")
    
    async def lookup(self, email: str) -> EmailIntelligence:
        """
        Perform comprehensive email lookup.
        
        Args:
            email: Email address to investigate
        
        Returns:
            EmailIntelligence report
        """
        email = email.lower().strip()
        intel = EmailIntelligence(email=email)
        
        logger.info(f"Starting email lookup: {email[:3]}***@{email.split('@')[-1]}")
        
        # Validate format
        intel.valid_format = bool(self.EMAIL_PATTERN.match(email))
        if not intel.valid_format:
            intel.risk_factors.append("Invalid email format")
            logger.warn(f"Invalid email format: {email}")
            return intel
        
        # Extract domain
        intel.domain = email.split('@')[1]
        
        # Check if disposable
        intel.disposable = intel.domain in DisposableEmailDomains.DOMAINS
        if intel.disposable:
            intel.risk_factors.append("Disposable email domain")
            intel.risk_level = RiskLevel.MEDIUM
        
        # Check domain MX records
        await self._check_mx_records(intel)
        
        # Check for breaches (simulated)
        await self._check_breaches(intel)
        
        # Get Gravatar
        intel.gravatar_url = self._get_gravatar_url(email)
        
        # Calculate risk level
        self._calculate_risk(intel)
        
        logger.verbose(f"Email lookup completed", {
            "email": email[:3] + "***",
            "risk_level": intel.risk_level.value,
            "breaches": len(intel.breaches)
        })
        
        return intel
    
    async def _check_mx_records(self, intel: EmailIntelligence):
        """Check domain MX records to verify email deliverability."""
        try:
            resolver = dns.resolver.Resolver()
            resolver.timeout = 5
            resolver.lifetime = 5
            
            answers = resolver.resolve(intel.domain, 'MX')
            intel.mx_records = [str(r.exchange) for r in answers]
            intel.domain_exists = True
            
            logger.debug(f"MX records found for {intel.domain}", {
                "count": len(intel.mx_records)
            })
            
        except dns.resolver.NXDOMAIN:
            intel.domain_exists = False
            intel.risk_factors.append("Domain does not exist")
            logger.warn(f"Domain does not exist: {intel.domain}")
            
        except Exception as e:
            logger.debug(f"MX lookup error: {e}")
    
    async def _check_breaches(self, intel: EmailIntelligence):
        """
        Check email against breach databases.
        
        Note: In production, integrate with HaveIBeenPwned API.
        This is a simulation for demo purposes.
        """
        # Simulate breach lookup
        # In production, call HIBP API:
        # https://haveibeenpwned.com/API/v3/breachedaccount/{email}
        
        # Simulated breaches for demo
        sample_breaches = [
            BreachRecord(
                name="LinkedIn",
                date="2012-05-05",
                records_count=164611595,
                data_types=["Emails", "Passwords"]
            ),
            BreachRecord(
                name="Adobe",
                date="2013-10-04",
                records_count=152445165,
                data_types=["Emails", "Passwords", "Usernames"]
            )
        ]
        
        # Check if email domain suggests potential exposure
        common_domains = ['gmail.com', 'yahoo.com', 'hotmail.com', 'outlook.com']
        if intel.domain in common_domains:
            # Simulate 30% chance of breach for common domains
            if hash(intel.email) % 10 < 3:
                intel.breaches = sample_breaches[:1]
                intel.risk_factors.append("Found in data breach")
        
        if intel.breaches:
            logger.threat_detected(
                threat_type="credential_exposure",
                severity="HIGH",
                domain=intel.domain,
                source="Breach Database",
                details={"breach_count": len(intel.breaches)}
            )
    
    def _get_gravatar_url(self, email: str) -> str:
        """Generate Gravatar URL from email."""
        email_hash = hashlib.md5(email.lower().encode()).hexdigest()
        return f"https://www.gravatar.com/avatar/{email_hash}?d=404"
    
    def _calculate_risk(self, intel: EmailIntelligence):
        """Calculate overall risk level based on findings."""
        risk_score = 0
        
        if not intel.valid_format:
            risk_score += 20
        if intel.disposable:
            risk_score += 30
        if not intel.domain_exists:
            risk_score += 40
        if intel.breaches:
            risk_score += 20 * len(intel.breaches)
        
        if risk_score >= 70:
            intel.risk_level = RiskLevel.CRITICAL
        elif risk_score >= 50:
            intel.risk_level = RiskLevel.HIGH
        elif risk_score >= 30:
            intel.risk_level = RiskLevel.MEDIUM
        elif risk_score >= 10:
            intel.risk_level = RiskLevel.LOW
        else:
            intel.risk_level = RiskLevel.INFO


# ============================================================================
# PHONE INTELLIGENCE
# ============================================================================

class PhoneLookup:
    """
    Phone number intelligence gathering.
    
    Features:
    - Number validation and formatting
    - Country/region detection
    - Carrier lookup
    - Line type identification (mobile/landline/VoIP)
    """
    
    # Country code patterns
    COUNTRY_CODES = {
        '1': ('US/CA', 'United States/Canada'),
        '44': ('GB', 'United Kingdom'),
        '33': ('FR', 'France'),
        '49': ('DE', 'Germany'),
        '81': ('JP', 'Japan'),
        '86': ('CN', 'China'),
        '91': ('IN', 'India'),
        '61': ('AU', 'Australia'),
        '7': ('RU', 'Russia'),
        '55': ('BR', 'Brazil'),
    }
    
    def __init__(self):
        logger.info("PhoneLookup initialized")
    
    async def lookup(self, number: str) -> PhoneIntelligence:
        """
        Perform phone number lookup.
        
        Args:
            number: Phone number to investigate
        
        Returns:
            PhoneIntelligence report
        """
        # Clean number
        cleaned = re.sub(r'[^\d+]', '', number)
        intel = PhoneIntelligence(number=number)
        
        logger.info(f"Starting phone lookup: {cleaned[:4]}****")
        
        # Basic validation
        if len(cleaned) < 10:
            intel.risk_factors.append("Number too short")
            return intel
        
        intel.valid = True
        
        # Detect country code
        self._detect_country(intel, cleaned)
        
        # Simulate carrier lookup
        self._lookup_carrier(intel, cleaned)
        
        # Simulate line type detection
        self._detect_line_type(intel, cleaned)
        
        logger.verbose(f"Phone lookup completed", {
            "country": intel.country_code,
            "carrier": intel.carrier,
            "line_type": intel.line_type
        })
        
        return intel
    
    def _detect_country(self, intel: PhoneIntelligence, number: str):
        """Detect country from phone number."""
        # Handle + prefix
        if number.startswith('+'):
            number = number[1:]
        
        # Try to match country codes
        for code, (iso, name) in self.COUNTRY_CODES.items():
            if number.startswith(code):
                intel.country_code = iso
                intel.country_name = name
                break
        
        # Default to US if starts with area code
        if not intel.country_code and len(number) == 10:
            intel.country_code = "US"
            intel.country_name = "United States"
    
    def _lookup_carrier(self, intel: PhoneIntelligence, number: str):
        """Simulate carrier lookup."""
        # In production, use services like Twilio Lookup API
        carriers = ["AT&T", "Verizon", "T-Mobile", "Sprint", "Unknown"]
        # Simple hash-based simulation
        intel.carrier = carriers[hash(number) % len(carriers)]
    
    def _detect_line_type(self, intel: PhoneIntelligence, number: str):
        """Simulate line type detection."""
        # In production, use carrier lookup services
        line_types = ["mobile", "landline", "voip", "toll-free"]
        
        # Simple pattern-based heuristic
        if number.startswith(('800', '888', '877', '866', '855')):
            intel.line_type = "toll-free"
        else:
            intel.line_type = line_types[hash(number) % 3]


# ============================================================================
# USERNAME ENUMERATION
# ============================================================================

class UsernameLookup:
    """
    Username enumeration across social platforms.
    
    Features:
    - Multi-platform username search
    - Concurrent checking
    - Profile URL discovery
    """
    
    def __init__(self, max_concurrent: int = 10, timeout: int = 10):
        """
        Initialize username lookup.
        
        Args:
            max_concurrent: Maximum concurrent requests
            timeout: Request timeout in seconds
        """
        self.max_concurrent = max_concurrent
        self.timeout = timeout
        logger.info("UsernameLookup initialized")
    
    async def lookup(
        self,
        username: str,
        platforms: Optional[List[str]] = None
    ) -> UsernameIntelligence:
        """
        Enumerate username across platforms.
        
        Args:
            username: Username to search
            platforms: Specific platforms to check (None = all)
        
        Returns:
            UsernameIntelligence report
        """
        intel = UsernameIntelligence(username=username)
        
        # Select platforms
        target_platforms = platforms or list(SocialPlatforms.PLATFORMS.keys())
        
        logger.info(f"Starting username enumeration: {username}", {
            "platforms": len(target_platforms)
        })
        
        # Create semaphore for rate limiting
        semaphore = asyncio.Semaphore(self.max_concurrent)
        
        async def check_platform(platform: str) -> Optional[Dict[str, str]]:
            """Check if username exists on platform."""
            async with semaphore:
                url_template = SocialPlatforms.PLATFORMS.get(platform)
                if not url_template:
                    return None
                
                url = url_template.format(username=username)
                
                try:
                    async with aiohttp.ClientSession() as session:
                        async with session.get(
                            url,
                            timeout=aiohttp.ClientTimeout(total=self.timeout),
                            allow_redirects=False
                        ) as response:
                            # Most platforms return 200 if user exists
                            # 404 or redirect usually means not found
                            if response.status == 200:
                                return {"platform": platform, "url": url}
                            
                except Exception as e:
                    logger.debug(f"Error checking {platform}: {e}")
                
                return None
        
        # Check all platforms concurrently
        tasks = [check_platform(p) for p in target_platforms]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for result in results:
            if isinstance(result, dict):
                intel.found_platforms.append(result)
            elif result is None:
                pass  # Not found
        
        # Generate possible emails
        intel.possible_emails = self._generate_possible_emails(username)
        
        logger.verbose(f"Username enumeration completed", {
            "username": username,
            "found": len(intel.found_platforms),
            "checked": len(target_platforms)
        })
        
        return intel
    
    def _generate_possible_emails(self, username: str) -> List[str]:
        """Generate possible email addresses from username."""
        common_domains = ['gmail.com', 'yahoo.com', 'outlook.com', 'protonmail.com']
        return [f"{username}@{domain}" for domain in common_domains]


# ============================================================================
# DOMAIN INTELLIGENCE
# ============================================================================

class DomainLookup:
    """
    Domain reconnaissance and intelligence.
    
    Features:
    - WHOIS lookup
    - DNS enumeration
    - Subdomain discovery
    - Technology detection
    """
    
    def __init__(self):
        logger.info("DomainLookup initialized")
    
    async def lookup(self, domain: str) -> DomainIntelligence:
        """
        Perform domain reconnaissance.
        
        Args:
            domain: Domain to investigate
        
        Returns:
            DomainIntelligence report
        """
        # Clean domain
        domain = domain.lower().strip()
        if domain.startswith(('http://', 'https://')):
            domain = urlparse(domain).netloc
        
        intel = DomainIntelligence(domain=domain)
        
        logger.info(f"Starting domain lookup: {domain}")
        
        # WHOIS lookup
        await self._whois_lookup(intel)
        
        # DNS enumeration
        await self._dns_lookup(intel)
        
        # Subdomain enumeration
        await self._subdomain_enum(intel)
        
        logger.verbose(f"Domain lookup completed", {
            "domain": domain,
            "registered": intel.registered,
            "subdomains": len(intel.subdomains)
        })
        
        return intel
    
    async def _whois_lookup(self, intel: DomainIntelligence):
        """Perform WHOIS lookup. No-op if python-whois is not installed."""
        if not WHOIS_AVAILABLE or whois is None:
            logger.debug("WHOIS lookup skipped (python-whois not installed)")
            return
        try:
            w = whois.whois(intel.domain)
            
            intel.registered = w.domain_name is not None
            intel.registrar = w.registrar
            
            if w.creation_date:
                date = w.creation_date
                if isinstance(date, list):
                    date = date[0]
                intel.creation_date = date.isoformat() if hasattr(date, 'isoformat') else str(date)
            
            if w.expiration_date:
                date = w.expiration_date
                if isinstance(date, list):
                    date = date[0]
                intel.expiration_date = date.isoformat() if hasattr(date, 'isoformat') else str(date)
            
            if w.name_servers:
                intel.nameservers = list(w.name_servers) if isinstance(w.name_servers, (list, set)) else [w.name_servers]
            
            logger.debug(f"WHOIS data retrieved for {intel.domain}")
            
        except Exception as e:
            logger.warn(f"WHOIS lookup failed: {e}")
    
    async def _dns_lookup(self, intel: DomainIntelligence):
        """Perform DNS record lookup."""
        resolver = dns.resolver.Resolver()
        resolver.timeout = 5
        resolver.lifetime = 5
        
        # A records
        try:
            answers = resolver.resolve(intel.domain, 'A')
            intel.a_records = [str(r) for r in answers]
        except Exception:
            pass
        
        # MX records
        try:
            answers = resolver.resolve(intel.domain, 'MX')
            intel.mx_records = [str(r.exchange) for r in answers]
        except Exception:
            pass
        
        # TXT records
        try:
            answers = resolver.resolve(intel.domain, 'TXT')
            intel.txt_records = [str(r) for r in answers]
        except Exception:
            pass
        
        logger.debug(f"DNS records for {intel.domain}", {
            "a_records": len(intel.a_records),
            "mx_records": len(intel.mx_records),
            "txt_records": len(intel.txt_records)
        })
    
    async def _subdomain_enum(self, intel: DomainIntelligence):
        """Enumerate common subdomains."""
        common_subdomains = [
            'www', 'mail', 'ftp', 'smtp', 'pop', 'imap', 'webmail',
            'admin', 'api', 'dev', 'staging', 'test', 'beta',
            'blog', 'shop', 'store', 'app', 'mobile', 'm',
            'cdn', 'static', 'media', 'assets', 'images',
            'vpn', 'remote', 'secure', 'portal', 'gateway',
            'ns1', 'ns2', 'dns', 'dns1', 'dns2'
        ]
        
        resolver = dns.resolver.Resolver()
        resolver.timeout = 2
        resolver.lifetime = 2
        
        found_subdomains = []
        
        for sub in common_subdomains:
            subdomain = f"{sub}.{intel.domain}"
            try:
                resolver.resolve(subdomain, 'A')
                found_subdomains.append(subdomain)
            except Exception:
                pass
        
        intel.subdomains = found_subdomains
        
        if found_subdomains:
            logger.verbose(f"Subdomains found for {intel.domain}", {
                "count": len(found_subdomains)
            })


# ============================================================================
# UNIFIED OSINT ENGINE
# ============================================================================

class OSINTEngine:
    """
    Unified OSINT lookup engine.
    
    Provides a single interface for all OSINT capabilities:
    - Email intelligence
    - Phone intelligence
    - Username enumeration
    - Domain reconnaissance
    """
    
    def __init__(
        self,
        hibp_api_key: Optional[str] = None,
        max_concurrent: int = 10
    ):
        """
        Initialize OSINT engine.
        
        Args:
            hibp_api_key: HaveIBeenPwned API key
            max_concurrent: Maximum concurrent requests
        """
        self.email_lookup = EmailLookup(hibp_api_key)
        self.phone_lookup = PhoneLookup()
        self.username_lookup = UsernameLookup(max_concurrent)
        self.domain_lookup = DomainLookup()
        
        logger.info("OSINTEngine initialized", {
            "modules": ["email", "phone", "username", "domain"]
        })
    
    async def investigate_email(self, email: str) -> EmailIntelligence:
        """Investigate email address."""
        return await self.email_lookup.lookup(email)
    
    async def investigate_phone(self, number: str) -> PhoneIntelligence:
        """Investigate phone number."""
        return await self.phone_lookup.lookup(number)
    
    async def investigate_username(
        self,
        username: str,
        platforms: Optional[List[str]] = None
    ) -> UsernameIntelligence:
        """Investigate username across platforms."""
        return await self.username_lookup.lookup(username, platforms)
    
    async def investigate_domain(self, domain: str) -> DomainIntelligence:
        """Investigate domain."""
        return await self.domain_lookup.lookup(domain)
    
    async def full_investigation(
        self,
        email: Optional[str] = None,
        phone: Optional[str] = None,
        username: Optional[str] = None,
        domain: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        Perform full investigation across all provided identifiers.
        
        Args:
            email: Email address to investigate
            phone: Phone number to investigate
            username: Username to investigate
            domain: Domain to investigate
        
        Returns:
            Combined investigation report
        """
        results = {}
        tasks = []
        
        if email:
            tasks.append(('email', self.investigate_email(email)))
        if phone:
            tasks.append(('phone', self.investigate_phone(phone)))
        if username:
            tasks.append(('username', self.investigate_username(username)))
        if domain:
            tasks.append(('domain', self.investigate_domain(domain)))
        
        # Run all investigations concurrently
        if tasks:
            for name, coro in tasks:
                try:
                    results[name] = await coro
                except Exception as e:
                    logger.error(f"Investigation failed for {name}: {e}")
                    results[name] = {"error": str(e)}
        
        return {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "results": {
                k: v.to_dict() if hasattr(v, 'to_dict') else v
                for k, v in results.items()
            }
        }
    
    def export_report(
        self,
        results: Dict[str, Any],
        output_path: Path,
        format: str = "json"
    ):
        """Export investigation results."""
        output_path = Path(output_path)
        
        if format == "json":
            with open(output_path, 'w') as f:
                json.dump(results, f, indent=2, default=str)
        
        logger.info(f"Report exported: {output_path}")


# ============================================================================
# DEMO & TESTING
# ============================================================================

async def demo_osint():
    """Demonstrate OSINT capabilities."""
    print("=" * 80)
    print("ShadowHunter OSINT Engine Demo")
    print("=" * 80)
    
    engine = OSINTEngine()
    
    # Demo 1: Email lookup
    print("\n📧 Demo 1: Email Investigation")
    print("-" * 60)
    
    email_intel = await engine.investigate_email("test@example.com")
    print(f"   Email: {email_intel.email}")
    print(f"   Valid format: {email_intel.valid_format}")
    print(f"   Disposable: {email_intel.disposable}")
    print(f"   Domain exists: {email_intel.domain_exists}")
    print(f"   MX records: {len(email_intel.mx_records)}")
    print(f"   Breaches: {len(email_intel.breaches)}")
    print(f"   Risk level: {email_intel.risk_level.value}")
    
    # Demo 2: Phone lookup
    print("\n📱 Demo 2: Phone Investigation")
    print("-" * 60)
    
    phone_intel = await engine.investigate_phone("+1-555-123-4567")
    print(f"   Number: {phone_intel.number}")
    print(f"   Valid: {phone_intel.valid}")
    print(f"   Country: {phone_intel.country_name}")
    print(f"   Carrier: {phone_intel.carrier}")
    print(f"   Line type: {phone_intel.line_type}")
    
    # Demo 3: Username enumeration
    print("\n👤 Demo 3: Username Enumeration")
    print("-" * 60)
    
    # Limited platforms for demo speed
    username_intel = await engine.investigate_username(
        "johndoe",
        platforms=["github", "twitter", "linkedin"]
    )
    print(f"   Username: {username_intel.username}")
    print(f"   Platforms found: {len(username_intel.found_platforms)}")
    for profile in username_intel.found_platforms:
        print(f"      - {profile['platform']}: {profile['url']}")
    
    # Demo 4: Domain lookup
    print("\n🌐 Demo 4: Domain Investigation")
    print("-" * 60)
    
    domain_intel = await engine.investigate_domain("google.com")
    print(f"   Domain: {domain_intel.domain}")
    print(f"   Registered: {domain_intel.registered}")
    print(f"   Registrar: {domain_intel.registrar}")
    print(f"   A records: {len(domain_intel.a_records)}")
    print(f"   MX records: {len(domain_intel.mx_records)}")
    print(f"   Subdomains: {len(domain_intel.subdomains)}")
    
    print("\n" + "=" * 80)
    print("✅ OSINT Demo completed!")
    print("=" * 80)


if __name__ == "__main__":
    asyncio.run(demo_osint())