engines/osint.py

"""
OSINT engine for comprehensive information gathering.
"""
from typing import Dict, List, Any, Optional
import asyncio
import json
from dataclasses import dataclass
import holehe.core as holehe
import subprocess
import tempfile
import os
import face_recognition
import numpy as np
from PIL import Image
import io
import requests
from geopy.geocoders import Nominatim
from geopy.exc import GeocoderTimedOut
import whois
from datetime import datetime
from tenacity import retry, stop_after_attempt, wait_exponential

@dataclass
class PersonInfo:
    name: str
    age: Optional[int] = None
    location: Optional[str] = None
    gender: Optional[str] = None
    social_profiles: List[Dict[str, str]] = None
    images: List[str] = None
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "age": self.age,
            "location": self.location,
            "gender": self.gender,
            "social_profiles": self.social_profiles or [],
            "images": self.images or []
        }

class OSINTEngine:
    def __init__(self):
        self.geolocator = Nominatim(user_agent="intelligent_search_engine")
        self.known_platforms = [
            "Twitter", "Instagram", "Facebook", "LinkedIn", "GitHub",
            "Reddit", "YouTube", "TikTok", "Pinterest", "Snapchat",
            "Twitch", "Medium", "Dev.to", "Stack Overflow"
        ]
    
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    async def search_username(self, username: str) -> Dict[str, Any]:
        """Search for username across multiple platforms."""
        results = {
            "username": username,
            "found_on": []
        }
        
        # Create a temporary file for sherlock results
        with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.txt') as tmp:
            tmp_path = tmp.name
        
        try:
            # Run sherlock as a subprocess
            process = subprocess.Popen(
                ["sherlock", username, "--output", tmp_path],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )
            stdout, stderr = process.communicate()
            
            # Read results from the temporary file
            if os.path.exists(tmp_path):
                with open(tmp_path, 'r') as f:
                    for line in f:
                        if "[+]" in line:  # Found profile
                            platform = line.split("[+]")[1].split(":")[0].strip()
                            url = line.split(":")[-1].strip()
                            results["found_on"].append({
                                "platform": platform,
                                "url": url
                            })
                        elif "[-]" in line:  # Not found
                            platform = line.split("[-]")[1].split(":")[0].strip()
                            results["found_on"].append({
                                "platform": platform,
                                "url": ""
                            })
                
                # Clean up temp file
                os.unlink(tmp_path)
        except Exception as e:
            print(f"Error running sherlock: {e}")
        
        # Use holehe for email-based search
        email = f"{username}@gmail.com"  # Example email
        holehe_results = await holehe.check_email(email)
        
        # Combine results
        for platform, data in holehe_results.items():
            if data.get("exists", False):
                results["found_on"].append({
                    "platform": platform,
                    "url": data.get("url", ""),
                    "confidence": data.get("confidence", "high")
                })
        
        return results
    
    async def search_person(self, name: str, location: Optional[str] = None, 
                          age: Optional[int] = None, gender: Optional[str] = None) -> PersonInfo:
        """Search for information about a person."""
        person = PersonInfo(
            name=name,
            age=age,
            location=location,
            gender=gender
        )
        
        # Initialize social profiles list
        person.social_profiles = []
        
        # Search for social media profiles
        username_variants = [
            name.replace(" ", ""),
            name.replace(" ", "_"),
            name.replace(" ", "."),
            name.lower().replace(" ", "")
        ]
        
        for username in username_variants:
            results = await self.search_username(username)
            person.social_profiles.extend(results.get("found_on", []))
        
        return person
    
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    async def analyze_image(self, image_data: bytes) -> Dict[str, Any]:
        """Analyze an image for faces and other identifiable information."""
        try:
            # Load image
            image = face_recognition.load_image_file(io.BytesIO(image_data))
            
            # Detect faces
            face_locations = face_recognition.face_locations(image)
            face_encodings = face_recognition.face_encodings(image, face_locations)
            
            results = {
                "faces_found": len(face_locations),
                "faces": []
            }
            
            # Analyze each face
            for i, (face_encoding, face_location) in enumerate(zip(face_encodings, face_locations)):
                face_data = {
                    "location": {
                        "top": face_location[0],
                        "right": face_location[1],
                        "bottom": face_location[2],
                        "left": face_location[3]
                    }
                }
                results["faces"].append(face_data)
            
            return results
        except Exception as e:
            return {"error": str(e)}
    
    async def search_location(self, location: str) -> Dict[str, Any]:
        """Gather information about a location."""
        try:
            # Geocode the location
            location_data = self.geolocator.geocode(location, timeout=10)
            
            if not location_data:
                return {"error": "Location not found"}
            
            return {
                "address": location_data.address,
                "latitude": location_data.latitude,
                "longitude": location_data.longitude,
                "raw": location_data.raw
            }
        except GeocoderTimedOut:
            return {"error": "Geocoding service timed out"}
        except Exception as e:
            return {"error": str(e)}
    
    async def analyze_domain(self, domain: str) -> Dict[str, Any]:
        """Analyze a domain for WHOIS and other information."""
        try:
            w = whois.whois(domain)
            return {
                "registrar": w.registrar,
                "creation_date": w.creation_date,
                "expiration_date": w.expiration_date,
                "last_updated": w.updated_date,
                "status": w.status,
                "name_servers": w.name_servers
            }
        except Exception as e:
            return {"error": str(e)}