Richard/eirescope
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

add OSINT modules: username search, email enrichment, phone analysis

Browse source
This commit is contained in:
authentik Default Admin 2026-02-20 22:30:00 +00:00
parent d1561f1120
commit faed675969
3 changed files with 588 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

236
eirescope/modules/email_module.py Normal file
View file

@ -0,0 +1,236 @@
"""Email OSINT Module — Email enrichment, breach checks, and domain analysis."""
import re
import socket
import logging
from typing import List, Dict, Optional
from eirescope.core.entity import Entity, EntityType, Investigation
from eirescope.modules.base import BaseOSINTModule
from eirescope.utils.http_client import OSINTHTTPClient
logger = logging.getLogger("eirescope.modules.email")
class EmailModule(BaseOSINTModule):
"""Enrich email addresses with breach data, domain info, and associated accounts."""
name = "Email Enrichment"
description = "Analyze email: validate domain, check breaches (HIBP), extract domain info, find associated usernames"
supported_entity_types = [EntityType.EMAIL]
requires_api_key = False # Core features work without API key; HIBP needs key for full results
api_key_name = "HIBP_API_KEY"
icon = "mail"
def __init__(self, config=None):
super().__init__(config)
self.http = OSINTHTTPClient(timeout=10, max_retries=2, rate_limit=0.5)
def execute(self, entity: Entity, investigation: Investigation) -> List[Entity]:
"""Run email enrichment pipeline."""
email = entity.value.lower().strip()
logger.info(f"Enriching email: {email}")
found_entities = []
local_part, domain = email.split("@", 1)
# 1. Extract username from email local part
username_entity = Entity(
entity_type=EntityType.USERNAME,
value=local_part,
source_module=self.name,
confidence=0.7,
metadata={"derived_from": email, "note": "Username extracted from email local part"},
)
added = investigation.add_entity(username_entity)
investigation.add_relationship(
source_id=entity.id,
target_id=added.id,
rel_type="email_contains_username",
confidence=0.7,
)
found_entities.append(added)
# 2. Extract and enrich domain
domain_entity = Entity(
entity_type=EntityType.DOMAIN,
value=domain,
source_module=self.name,
confidence=1.0,
metadata={"derived_from": email},
)
added_domain = investigation.add_entity(domain_entity)
investigation.add_relationship(
source_id=entity.id,
target_id=added_domain.id,
rel_type="email_hosted_on",
confidence=1.0,
)
found_entities.append(added_domain)
# 3. Validate domain via MX records
mx_info = self._check_mx_records(domain)
entity.metadata["mx_records"] = mx_info
entity.metadata["domain"] = domain
entity.metadata["local_part"] = local_part
entity.metadata["is_valid_domain"] = len(mx_info) > 0
# 4. Detect email provider
provider = self._detect_provider(domain, mx_info)
entity.metadata["email_provider"] = provider
# 5. Check for known disposable email domains
entity.metadata["is_disposable"] = self._is_disposable_domain(domain)
# 6. Check HaveIBeenPwned (if API key available)
breach_results = self._check_breaches(email)
if breach_results:
entity.metadata["breaches"] = breach_results
entity.metadata["breach_count"] = len(breach_results)
for breach in breach_results:
breach_entity = Entity(
entity_type=EntityType.BREACH,
value=breach.get("name", "Unknown Breach"),
source_module=self.name,
confidence=0.95,
metadata=breach,
)
added_breach = investigation.add_entity(breach_entity)
investigation.add_relationship(
source_id=entity.id,
target_id=added_breach.id,
rel_type="found_in_breach",
confidence=0.95,
evidence={"breach_name": breach.get("name"), "breach_date": breach.get("date")},
)
found_entities.append(added_breach)
# 7. Check Gravatar for profile
gravatar_info = self._check_gravatar(email)
if gravatar_info:
entity.metadata["gravatar"] = gravatar_info
logger.info(f"Email enrichment complete: {len(found_entities)} entities discovered")
return found_entities
def _check_mx_records(self, domain: str) -> List[Dict]:
"""Check MX records for the email domain."""
try:
import subprocess
result = subprocess.run(
["dig", "+short", "MX", domain],
capture_output=True, text=True, timeout=10,
)
if result.returncode == 0 and result.stdout.strip():
records = []
for line in result.stdout.strip().split("\n"):
parts = line.strip().split()
if len(parts) >= 2:
records.append({
"priority": int(parts[0]),
"server": parts[1].rstrip("."),
})
return sorted(records, key=lambda x: x["priority"])
except Exception as e:
logger.debug(f"MX lookup failed for {domain}: {e}")
# Fallback: try socket
try:
socket.getaddrinfo(domain, 25)
return [{"priority": 0, "server": domain, "note": "fallback check"}]
except Exception:
pass
return []
def _detect_provider(self, domain: str, mx_records: List[Dict]) -> str:
"""Detect email provider from domain or MX records."""
domain_lower = domain.lower()
mx_str = " ".join([r.get("server", "") for r in mx_records]).lower()
provider_map = {
"gmail.com": "Google (Gmail)",
"googlemail.com": "Google (Gmail)",
"outlook.com": "Microsoft (Outlook)",
"hotmail.com": "Microsoft (Hotmail)",
"live.com": "Microsoft (Live)",
"yahoo.com": "Yahoo",
"protonmail.com": "ProtonMail",
"proton.me": "ProtonMail",
"icloud.com": "Apple (iCloud)",
"me.com": "Apple",
"aol.com": "AOL",
"zoho.com": "Zoho",
}
if domain_lower in provider_map:
return provider_map[domain_lower]
if "google" in mx_str or "gmail" in mx_str:
return "Google Workspace"
if "outlook" in mx_str or "microsoft" in mx_str:
return "Microsoft 365"
if "protonmail" in mx_str:
return "ProtonMail"
if "zoho" in mx_str:
return "Zoho"
return "Custom/Unknown"
def _is_disposable_domain(self, domain: str) -> bool:
"""Check if domain is a known disposable/temporary email provider."""
disposable_domains = {
"tempmail.com", "guerrillamail.com", "mailinator.com",
"throwaway.email", "temp-mail.org", "fakeinbox.com",
"sharklasers.com", "guerrillamailblock.com", "grr.la",
"dispostable.com", "yopmail.com", "trashmail.com",
"maildrop.cc", "10minutemail.com", "tempail.com",
"burnermail.io", "mailnesia.com", "tempr.email",
}
return domain.lower() in disposable_domains
def _check_breaches(self, email: str) -> List[Dict]:
"""Check HaveIBeenPwned for data breaches (requires API key for full results)."""
if self.api_key:
try:
resp = self.http.get(
f"https://haveibeenpwned.com/api/v3/breachedaccount/{email}",
headers={
"hibp-api-key": self.api_key,
"User-Agent": "EireScope-OSINT",
},
)
if resp and resp.status_code == 200:
return resp.json()
elif resp and resp.status_code == 404:
return []
except Exception as e:
logger.debug(f"HIBP check failed: {e}")
# Free alternative: check breach directory (basic)
try:
resp = self.http.get(
f"https://api.xposedornot.com/v1/check-email/{email}"
)
if resp and resp.status_code == 200:
data = resp.json()
if "breaches" in data:
return [{"name": b, "source": "XposedOrNot"} for b in data["breaches"]]
except Exception as e:
logger.debug(f"Alternative breach check failed: {e}")
return []
def _check_gravatar(self, email: str) -> Optional[Dict]:
"""Check for Gravatar profile associated with email."""
import hashlib
email_hash = hashlib.md5(email.lower().encode()).hexdigest()
url = f"https://www.gravatar.com/avatar/{email_hash}?d=404"
try:
resp = self.http.head(url)
if resp and resp.status_code == 200:
return {
"has_gravatar": True,
"avatar_url": f"https://www.gravatar.com/avatar/{email_hash}",
"profile_url": f"https://gravatar.com/{email_hash}",
}
except Exception:
pass
return None

200
eirescope/modules/phone_module.py Normal file
View file

@ -0,0 +1,200 @@
"""Phone Number OSINT Module — Validation, carrier detection, geolocation."""
import re
import logging
from typing import List, Dict, Optional
from eirescope.core.entity import Entity, EntityType, Investigation
from eirescope.modules.base import BaseOSINTModule
logger = logging.getLogger("eirescope.modules.phone")
# Irish carrier prefixes (mobile)
IRISH_CARRIERS = {
"083": "Three Ireland",
"085": "Three Ireland",
"086": "Vodafone Ireland",
"087": "Vodafone Ireland",
"089": "Three Ireland",
"088": "Tesco Mobile / Three MVNO",
}
# Country codes
COUNTRY_CODES = {
"353": {"country": "Ireland", "iso": "IE", "name": "Éire"},
"44": {"country": "United Kingdom", "iso": "GB", "name": "UK"},
"1": {"country": "United States/Canada", "iso": "US/CA", "name": "North America"},
"33": {"country": "France", "iso": "FR", "name": "France"},
"49": {"country": "Germany", "iso": "DE", "name": "Germany"},
"34": {"country": "Spain", "iso": "ES", "name": "Spain"},
"39": {"country": "Italy", "iso": "IT", "name": "Italy"},
"31": {"country": "Netherlands", "iso": "NL", "name": "Netherlands"},
"32": {"country": "Belgium", "iso": "BE", "name": "Belgium"},
"48": {"country": "Poland", "iso": "PL", "name": "Poland"},
"351": {"country": "Portugal", "iso": "PT", "name": "Portugal"},
"45": {"country": "Denmark", "iso": "DK", "name": "Denmark"},
"46": {"country": "Sweden", "iso": "SE", "name": "Sweden"},
"47": {"country": "Norway", "iso": "NO", "name": "Norway"},
"358": {"country": "Finland", "iso": "FI", "name": "Finland"},
"91": {"country": "India", "iso": "IN", "name": "India"},
"86": {"country": "China", "iso": "CN", "name": "China"},
"81": {"country": "Japan", "iso": "JP", "name": "Japan"},
"55": {"country": "Brazil", "iso": "BR", "name": "Brazil"},
"61": {"country": "Australia", "iso": "AU", "name": "Australia"},
}
class PhoneModule(BaseOSINTModule):
"""Analyze and enrich phone number data."""
name = "Phone Number Analysis"
description = "Validate phone numbers, detect carrier (Irish focus), identify country and number type"
supported_entity_types = [EntityType.PHONE]
requires_api_key = False
icon = "phone"
def execute(self, entity: Entity, investigation: Investigation) -> List[Entity]:
"""Analyze phone number."""
phone = entity.value
logger.info(f"Analyzing phone number: {phone}")
found_entities = []
# Normalize
cleaned = re.sub(r"[\s\-\(\)\.]", "", phone)
if not cleaned.startswith("+"):
# Assume Irish if starts with 0
if cleaned.startswith("0"):
cleaned = "+353" + cleaned[1:]
else:
cleaned = "+" + cleaned
entity.metadata["normalized"] = cleaned
entity.metadata["original"] = phone
# Detect country
country_info = self._detect_country(cleaned)
entity.metadata["country"] = country_info
if country_info:
geo_entity = Entity(
entity_type=EntityType.GEO_LOCATION,
value=country_info.get("country", "Unknown"),
source_module=self.name,
confidence=0.9,
metadata=country_info,
)
added = investigation.add_entity(geo_entity)
investigation.add_relationship(
source_id=entity.id,
target_id=added.id,
rel_type="phone_registered_in",
confidence=0.9,
)
found_entities.append(added)
# Detect carrier (Irish numbers)
carrier_info = self._detect_irish_carrier(cleaned)
if carrier_info:
entity.metadata["carrier"] = carrier_info
carrier_entity = Entity(
entity_type=EntityType.CARRIER_INFO,
value=carrier_info["carrier"],
source_module=self.name,
confidence=0.8,
metadata=carrier_info,
)
added = investigation.add_entity(carrier_entity)
investigation.add_relationship(
source_id=entity.id,
target_id=added.id,
rel_type="phone_carrier_is",
confidence=0.8,
)
found_entities.append(added)
# Classify number type
number_type = self._classify_number_type(cleaned)
entity.metadata["number_type"] = number_type
# Format validation
entity.metadata["is_valid_format"] = self._validate_format(cleaned)
entity.metadata["e164_format"] = cleaned
# Irish-specific checks
if country_info and country_info.get("iso") == "IE":
entity.metadata["irish_analysis"] = self._analyze_irish_number(cleaned)
logger.info(f"Phone analysis complete: {len(found_entities)} entities discovered")
return found_entities
def _detect_country(self, phone: str) -> Optional[Dict]:
"""Detect country from phone number prefix."""
digits = phone.lstrip("+")
# Try 3-digit codes first, then 2-digit, then 1-digit
for length in [3, 2, 1]:
prefix = digits[:length]
if prefix in COUNTRY_CODES:
return COUNTRY_CODES[prefix].copy()
return None
def _detect_irish_carrier(self, phone: str) -> Optional[Dict]:
"""Detect Irish mobile carrier from phone prefix."""
# Convert +353 to 0-prefix format for carrier lookup
if phone.startswith("+353"):
local = "0" + phone[4:]
prefix = local[:3]
if prefix in IRISH_CARRIERS:
return {
"carrier": IRISH_CARRIERS[prefix],
"prefix": prefix,
"type": "mobile",
"country": "Ireland",
}
return None
def _classify_number_type(self, phone: str) -> str:
"""Classify number as mobile, landline, VoIP, toll-free, etc."""
if phone.startswith("+353"):
local = phone[4:]
if local.startswith("1"):
return "landline (Dublin)"
if local.startswith(("21", "22", "23", "24", "25", "26", "27", "28", "29")):
return "landline (Munster)"
if local.startswith(("41", "42", "43", "44", "45", "46", "47", "49")):
return "landline (Leinster/Ulster)"
if local.startswith(("51", "52", "53", "54", "56", "57", "58", "59")):
return "landline (South-East)"
if local.startswith(("61", "62", "63", "64", "65", "66", "67", "68", "69")):
return "landline (Mid-West/Kerry)"
if local.startswith(("71", "74", "76", "90", "91", "93", "94", "95", "96", "97", "98", "99")):
return "landline (West/North-West)"
if local.startswith(("83", "85", "86", "87", "89")):
return "mobile"
if local.startswith("1800"):
return "toll-free"
if local.startswith("1850") or local.startswith("1890"):
return "shared-cost"
return "unknown"
def _validate_format(self, phone: str) -> bool:
"""Validate E.164 format."""
return bool(re.match(r"^\+\d{7,15}$", phone))
def _analyze_irish_number(self, phone: str) -> Dict:
"""Additional analysis for Irish phone numbers."""
local = phone[4:] # Remove +353
analysis = {
"is_irish": True,
"local_number": "0" + local,
"international_format": phone,
}
# Check if it's a premium rate number
if local.startswith("15"):
analysis["warning"] = "Premium rate number"
analysis["risk_level"] = "high"
# Check for known VoIP ranges
if local.startswith("76"):
analysis["note"] = "VoIP number range — may be harder to trace"
analysis["is_voip"] = True
return analysis

152
eirescope/modules/username_module.py Normal file
View file

@ -0,0 +1,152 @@
"""Username OSINT Module — Search username across social platforms (Sherlock-like)."""
import logging
from typing import List
from concurrent.futures import ThreadPoolExecutor, as_completed
from eirescope.core.entity import Entity, EntityType, Investigation
from eirescope.modules.base import BaseOSINTModule
from eirescope.utils.http_client import OSINTHTTPClient
logger = logging.getLogger("eirescope.modules.username")
# Platforms to check: name, URL template ({} = username), error indicator
PLATFORMS = [
# Major Social Media
{"name": "GitHub", "url": "https://github.com/{}", "category": "development"},
{"name": "GitLab", "url": "https://gitlab.com/{}", "category": "development"},
{"name": "Twitter/X", "url": "https://x.com/{}", "category": "social"},
{"name": "Instagram", "url": "https://www.instagram.com/{}/", "category": "social"},
{"name": "Reddit", "url": "https://www.reddit.com/user/{}", "category": "social"},
{"name": "TikTok", "url": "https://www.tiktok.com/@{}", "category": "social"},
{"name": "YouTube", "url": "https://www.youtube.com/@{}", "category": "social"},
{"name": "Twitch", "url": "https://www.twitch.tv/{}", "category": "social"},
{"name": "Pinterest", "url": "https://www.pinterest.com/{}/", "category": "social"},
{"name": "Tumblr", "url": "https://{}.tumblr.com", "category": "social"},
# Professional
{"name": "LinkedIn", "url": "https://www.linkedin.com/in/{}", "category": "professional"},
{"name": "Medium", "url": "https://medium.com/@{}", "category": "professional"},
{"name": "Dev.to", "url": "https://dev.to/{}", "category": "development"},
{"name": "HackerNews", "url": "https://news.ycombinator.com/user?id={}", "category": "development"},
{"name": "StackOverflow", "url": "https://stackoverflow.com/users/?tab=accounts&SearchText={}", "category": "development"},
{"name": "Keybase", "url": "https://keybase.io/{}", "category": "security"},
# Communication
{"name": "Telegram", "url": "https://t.me/{}", "category": "communication"},
{"name": "Mastodon (social)", "url": "https://mastodon.social/@{}", "category": "social"},
# Content & Media
{"name": "SoundCloud", "url": "https://soundcloud.com/{}", "category": "media"},
{"name": "Spotify", "url": "https://open.spotify.com/user/{}", "category": "media"},
{"name": "Flickr", "url": "https://www.flickr.com/people/{}", "category": "media"},
{"name": "Vimeo", "url": "https://vimeo.com/{}", "category": "media"},
{"name": "Dailymotion", "url": "https://www.dailymotion.com/{}", "category": "media"},
# Gaming
{"name": "Steam Community", "url": "https://steamcommunity.com/id/{}", "category": "gaming"},
{"name": "Xbox Gamertag", "url": "https://xboxgamertag.com/search/{}", "category": "gaming"},
# Forums & Communities
{"name": "HackerOne", "url": "https://hackerone.com/{}", "category": "security"},
{"name": "Bugcrowd", "url": "https://bugcrowd.com/{}", "category": "security"},
{"name": "Gravatar", "url": "https://en.gravatar.com/{}", "category": "other"},
{"name": "About.me", "url": "https://about.me/{}", "category": "professional"},
{"name": "Behance", "url": "https://www.behance.net/{}", "category": "professional"},
{"name": "Dribbble", "url": "https://dribbble.com/{}", "category": "professional"},
# Tech & Code
{"name": "Replit", "url": "https://replit.com/@{}", "category": "development"},
{"name": "CodePen", "url": "https://codepen.io/{}", "category": "development"},
{"name": "npm", "url": "https://www.npmjs.com/~{}", "category": "development"},
{"name": "PyPI", "url": "https://pypi.org/user/{}/", "category": "development"},
{"name": "Docker Hub", "url": "https://hub.docker.com/u/{}", "category": "development"},
# News & Blogging
{"name": "Blogger", "url": "https://{}.blogspot.com", "category": "blogging"},
{"name": "WordPress", "url": "https://{}.wordpress.com", "category": "blogging"},
{"name": "Substack", "url": "https://{}.substack.com", "category": "blogging"},
# Irish / EU specific
{"name": "Boards.ie", "url": "https://www.boards.ie/member/{}", "category": "irish"},
]
class UsernameModule(BaseOSINTModule):
"""Search for a username across multiple social platforms."""
name = "Username Search"
description = "Search a username across 40+ social platforms and websites (Sherlock-like)"
supported_entity_types = [EntityType.USERNAME]
requires_api_key = False
icon = "user-search"
def __init__(self, config=None):
super().__init__(config)
self.http = OSINTHTTPClient(
timeout=config.get("timeout", 8) if config else 8,
max_retries=1,
rate_limit=0.1,
)
self.max_workers = 10
def _check_platform(self, platform: dict, username: str) -> dict:
"""Check if username exists on a single platform."""
url = platform["url"].format(username)
try:
exists = self.http.check_url_exists(url, timeout=6)
return {
"platform": platform["name"],
"url": url,
"exists": exists,
"category": platform["category"],
}
except Exception as e:
logger.debug(f"Error checking {platform['name']}: {e}")
return {
"platform": platform["name"],
"url": url,
"exists": False,
"category": platform["category"],
"error": str(e),
}
def execute(self, entity: Entity, investigation: Investigation) -> List[Entity]:
"""Search username across all platforms using thread pool."""
username = entity.value
logger.info(f"Searching username '{username}' across {len(PLATFORMS)} platforms")
found_entities = []
results = []
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {
executor.submit(self._check_platform, p, username): p
for p in PLATFORMS
}
for future in as_completed(futures):
result = future.result()
results.append(result)
if result["exists"]:
profile_entity = Entity(
entity_type=EntityType.SOCIAL_PROFILE,
value=result["url"],
source_module=self.name,
confidence=0.85,
metadata={
"platform": result["platform"],
"category": result["category"],
"username": username,
},
)
added = investigation.add_entity(profile_entity)
investigation.add_relationship(
source_id=entity.id,
target_id=added.id,
rel_type="has_profile_on",
confidence=0.85,
evidence={"url": result["url"], "platform": result["platform"]},
)
found_entities.append(added)
logger.info(f" [+] Found: {result['platform']} → {result['url']}")
# Store summary in the original entity metadata
entity.metadata["platforms_checked"] = len(PLATFORMS)
entity.metadata["profiles_found"] = len(found_entities)
entity.metadata["results_summary"] = [
{"platform": r["platform"], "url": r["url"], "found": r["exists"]}
for r in sorted(results, key=lambda x: x["platform"])
]
logger.info(f"Username search complete: {len(found_entities)}/{len(PLATFORMS)} platforms found")
return found_entities