Update engines/osint.py

engines/osint.py

@@ -1,202 +1,189 @@
 """
-OSINT engine for comprehensive information gathering.
+OSINT engine for username and person search.
 """
 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
+from bs4 import BeautifulSoup
 import whois
-from datetime import datetime
+from holehe.core import *
+from geopy.geocoders import Nominatim
 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 []
-        }
+from duckduckgo_search import DDGS
 
 class OSINTEngine:
     def __init__(self):
-        self.geolocator = Nominatim(user_agent="intelligent_search_engine")
+        self.geolocator = Nominatim(user_agent="ise_search")
+        self.holehe_modules = import_submodules("holehe.modules")
         self.known_platforms = [
-            "Twitter", "Instagram", "Facebook", "LinkedIn", "GitHub",
-            "Reddit", "YouTube", "TikTok", "Pinterest", "Snapchat",
-            "Twitch", "Medium", "Dev.to", "Stack Overflow"
+            "twitter.com", "facebook.com", "instagram.com", "linkedin.com",
+            "github.com", "youtube.com", "reddit.com", "pinterest.com",
+            "medium.com", "tumblr.com", "flickr.com", "vimeo.com"
         ]
     
-    @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."""
+        """Search for username across platforms."""
         results = {
-            "username": username,
-            "found_on": []
+            "found": [],
+            "not_found": [],
+            "error": []
         }
         
-        # Create a temporary file for sherlock results
-        with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.txt') as tmp:
-            tmp_path = tmp.name
+        # Manual platform check
+        for platform in self.known_platforms:
+            try:
+                url = f"https://{platform}/{username}"
+                response = requests.head(url, timeout=5, allow_redirects=True)
+                if response.status_code == 200:
+                    results["found"].append({
+                        "platform": platform.split(".")[0].title(),
+                        "url": url
+                    })
+                else:
+                    results["not_found"].append(platform.split(".")[0].title())
+            except Exception as e:
+                results["error"].append({
+                    "platform": platform,
+                    "error": str(e)
+                })
         
+        # Run holehe checks
         try:
-            # Run sherlock as a subprocess
-            process = subprocess.Popen(
-                ["sherlock", username, "--output", tmp_path],
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE
-            )
-            stdout, stderr = process.communicate()
+            holehe_results = []
+            for module in self.holehe_modules:
+                try:
+                    check_func = getattr(module, "check")
+                    out = await check_func(username)
+                    if out and out.get("exists"):
+                        results["found"].append({
+                            "platform": out["name"],
+                            "url": out.get("url", ""),
+                            "email": out.get("email", "")
+                        })
+                except Exception as e:
+                    print(f"Error in holehe module {module.__name__}: {e}")
             
-            # 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)
+            results["holehe"] = holehe_results
         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")
-                })
+            print(f"Error running holehe: {e}")
         
         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."""
+    async def search_person(self, name: str, location: Optional[str] = None, age: Optional[int] = None) -> Dict[str, Any]:
+        """Search for person information."""
+        results = {
+            "basic_info": {},
+            "social_profiles": [],
+            "locations": [],
+            "possible_relatives": [],
+            "error": None
+        }
+        
         try:
-            # Load image
-            image = face_recognition.load_image_file(io.BytesIO(image_data))
+            # Geocode location if provided
+            if location:
+                try:
+                    loc = self.geolocator.geocode(location)
+                    if loc:
+                        results["locations"].append({
+                            "address": loc.address,
+                            "latitude": loc.latitude,
+                            "longitude": loc.longitude
+                        })
+                except Exception as e:
+                    print(f"Error geocoding location: {e}")
             
-            # Detect faces
-            face_locations = face_recognition.face_locations(image)
-            face_encodings = face_recognition.face_encodings(image, face_locations)
+            # Basic search query
+            search_query = f"{name}"
+            if location:
+                search_query += f" {location}"
+            if age:
+                search_query += f" {age} years old"
             
-            results = {
-                "faces_found": len(face_locations),
-                "faces": []
-            }
+            # Use DuckDuckGo for initial search
+            with DDGS() as ddgs:
+                search_results = [r for r in ddgs.text(search_query, max_results=10)]
             
-            # 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)
+            for result in search_results:
+                try:
+                    url = result["link"]
+                    # Check if URL is from a known social platform
+                    if any(platform in url.lower() for platform in self.known_platforms):
+                        platform = next(p for p in self.known_platforms if p in url.lower())
+                        results["social_profiles"].append({
+                            "platform": platform.split(".")[0].title(),
+                            "url": url,
+                            "title": result.get("title", "")
+                        })
+                except Exception as e:
+                    print(f"Error processing search result: {e}")
             
-            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)}
+            results["error"] = str(e)
+        
+        return results
     
-    async def analyze_domain(self, domain: str) -> Dict[str, Any]:
-        """Analyze a domain for WHOIS and other information."""
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
+    async def domain_lookup(self, domain: str) -> Dict[str, Any]:
+        """Perform WHOIS lookup for a domain."""
         try:
             w = whois.whois(domain)
             return {
+                "domain_name": w.domain_name,
                 "registrar": w.registrar,
                 "creation_date": w.creation_date,
                 "expiration_date": w.expiration_date,
-                "last_updated": w.updated_date,
+                "name_servers": w.name_servers,
                 "status": w.status,
-                "name_servers": w.name_servers
+                "emails": w.emails,
+                "dnssec": w.dnssec,
+                "name": w.name,
+                "org": w.org,
+                "address": w.address,
+                "city": w.city,
+                "state": w.state,
+                "zipcode": w.zipcode,
+                "country": w.country
             }
         except Exception as e:
             return {"error": str(e)}
+    
+    async def analyze_social_profile(self, url: str) -> Dict[str, Any]:
+        """Analyze a social media profile."""
+        results = {
+            "profile_info": {},
+            "recent_activity": [],
+            "connections": [],
+            "error": None
+        }
+        
+        try:
+            headers = {
+                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
+            }
+            response = requests.get(url, headers=headers, timeout=10)
+            response.raise_for_status()
+            
+            soup = BeautifulSoup(response.text, "html.parser")
+            
+            # Extract basic profile info
+            results["profile_info"]["title"] = soup.title.string if soup.title else None
+            
+            # Extract meta information
+            for meta in soup.find_all("meta"):
+                property = meta.get("property", "")
+                content = meta.get("content", "")
+                
+                if "og:title" in property:
+                    results["profile_info"]["og_title"] = content
+                elif "og:description" in property:
+                    results["profile_info"]["og_description"] = content
+                elif "og:image" in property:
+                    results["profile_info"]["og_image"] = content
+            
+        except Exception as e:
+            results["error"] = str(e)
+        
+        return results