add OSINT modules: IP recon, domain recon, social media discovery

eirescope/modules/domain_module.py

@@ -0,0 +1,228 @@
+"""Domain OSINT Module — DNS records, WHOIS, subdomain enumeration."""
+import re
+import socket
+import subprocess
+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.domain")
+
+
+class DomainModule(BaseOSINTModule):
+    """Domain reconnaissance — DNS, WHOIS, subdomain discovery."""
+
+    name = "Domain Recon"
+    description = "Analyze domain: DNS records (A, MX, NS, TXT), WHOIS registration, subdomain enumeration via crt.sh"
+    supported_entity_types = [EntityType.DOMAIN]
+    requires_api_key = False
+    icon = "globe"
+
+    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 domain reconnaissance."""
+        domain = entity.value.lower().strip().rstrip(".")
+        logger.info(f"Analyzing domain: {domain}")
+        found_entities = []
+
+        # 1. DNS A records → get IP addresses
+        a_records = self._dns_lookup(domain, "A")
+        entity.metadata["a_records"] = a_records
+        for ip in a_records:
+            ip_entity = Entity(
+                entity_type=EntityType.IP_ADDRESS,
+                value=ip,
+                source_module=self.name,
+                confidence=0.95,
+                metadata={"domain": domain, "record_type": "A"},
+            )
+            added = investigation.add_entity(ip_entity)
+            investigation.add_relationship(
+                source_id=entity.id,
+                target_id=added.id,
+                rel_type="domain_resolves_to",
+                confidence=0.95,
+            )
+            found_entities.append(added)
+
+        # 2. MX records
+        mx_records = self._dns_lookup(domain, "MX")
+        entity.metadata["mx_records"] = mx_records
+
+        # 3. NS records
+        ns_records = self._dns_lookup(domain, "NS")
+        entity.metadata["ns_records"] = ns_records
+
+        # 4. TXT records (SPF, DKIM, DMARC)
+        txt_records = self._dns_lookup(domain, "TXT")
+        entity.metadata["txt_records"] = txt_records
+        entity.metadata["spf"] = [r for r in txt_records if "spf" in r.lower()]
+        entity.metadata["dmarc"] = self._dns_lookup(f"_dmarc.{domain}", "TXT")
+
+        # 5. WHOIS
+        whois_data = self._whois_lookup(domain)
+        if whois_data:
+            entity.metadata["whois"] = whois_data
+            whois_entity = Entity(
+                entity_type=EntityType.WHOIS_INFO,
+                value=f"WHOIS: {domain}",
+                source_module=self.name,
+                confidence=0.9,
+                metadata=whois_data,
+            )
+            added = investigation.add_entity(whois_entity)
+            investigation.add_relationship(
+                source_id=entity.id,
+                target_id=added.id,
+                rel_type="has_whois_record",
+                confidence=0.9,
+            )
+            found_entities.append(added)
+
+            # Extract registrant email if available
+            if whois_data.get("registrant_email"):
+                email_entity = Entity(
+                    entity_type=EntityType.EMAIL,
+                    value=whois_data["registrant_email"],
+                    source_module=self.name,
+                    confidence=0.7,
+                    metadata={"source": "domain_whois", "domain": domain},
+                )
+                added = investigation.add_entity(email_entity)
+                investigation.add_relationship(
+                    source_id=entity.id,
+                    target_id=added.id,
+                    rel_type="domain_registered_by",
+                    confidence=0.7,
+                )
+                found_entities.append(added)
+
+        # 6. Subdomain enumeration via crt.sh
+        subdomains = self._enumerate_subdomains(domain)
+        entity.metadata["subdomains"] = subdomains
+        entity.metadata["subdomain_count"] = len(subdomains)
+        for sub in subdomains[:20]:  # Limit to top 20
+            sub_entity = Entity(
+                entity_type=EntityType.DOMAIN,
+                value=sub,
+                source_module=self.name,
+                confidence=0.8,
+                metadata={"parent_domain": domain, "source": "crt.sh"},
+            )
+            added = investigation.add_entity(sub_entity)
+            investigation.add_relationship(
+                source_id=entity.id,
+                target_id=added.id,
+                rel_type="has_subdomain",
+                confidence=0.8,
+            )
+            found_entities.append(added)
+
+        # 7. Security analysis
+        entity.metadata["security"] = {
+            "has_spf": len(entity.metadata.get("spf", [])) > 0,
+            "has_dmarc": len(entity.metadata.get("dmarc", [])) > 0,
+            "mx_count": len(mx_records),
+            "ns_count": len(ns_records),
+        }
+
+        logger.info(f"Domain analysis complete: {len(found_entities)} entities discovered")
+        return found_entities
+
+    def _dns_lookup(self, domain: str, record_type: str) -> List[str]:
+        """DNS lookup using dig command."""
+        try:
+            result = subprocess.run(
+                ["dig", "+short", record_type, 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"):
+                    val = line.strip().rstrip(".")
+                    if val and not val.startswith(";"):
+                        # For MX, extract just the server name
+                        if record_type == "MX":
+                            parts = val.split()
+                            if len(parts) >= 2:
+                                val = f"{parts[0]} {parts[1].rstrip('.')}"
+                        records.append(val)
+                return records
+        except Exception as e:
+            logger.debug(f"DNS {record_type} lookup failed for {domain}: {e}")
+
+        # Fallback for A records using socket
+        if record_type == "A":
+            try:
+                results = socket.getaddrinfo(domain, None, socket.AF_INET)
+                return list(set(r[4][0] for r in results))
+            except Exception:
+                pass
+        return []
+
+    def _whois_lookup(self, domain: str) -> Optional[Dict]:
+        """WHOIS lookup for domain."""
+        try:
+            result = subprocess.run(
+                ["whois", domain],
+                capture_output=True, text=True, timeout=15,
+            )
+            if result.returncode == 0 and result.stdout:
+                return self._parse_whois(result.stdout)
+        except FileNotFoundError:
+            logger.debug("whois command not found")
+        except Exception as e:
+            logger.debug(f"WHOIS lookup failed for {domain}: {e}")
+        return None
+
+    def _parse_whois(self, raw: str) -> Dict:
+        """Parse domain WHOIS output."""
+        data = {"raw": raw[:3000]}
+        patterns = {
+            "registrar": r"(?:Registrar|registrar):\s*(.+)",
+            "creation_date": r"(?:Creation Date|created):\s*(.+)",
+            "expiry_date": r"(?:Registry Expiry Date|Expiration Date|expires):\s*(.+)",
+            "updated_date": r"(?:Updated Date|last-modified):\s*(.+)",
+            "registrant_name": r"(?:Registrant Name|registrant):\s*(.+)",
+            "registrant_org": r"(?:Registrant Organization|org):\s*(.+)",
+            "registrant_email": r"(?:Registrant Email|e-mail):\s*(\S+@\S+)",
+            "registrant_country": r"(?:Registrant Country|country):\s*(\S+)",
+            "name_servers": r"(?:Name Server|nserver):\s*(\S+)",
+            "status": r"(?:Domain Status|status):\s*(.+)",
+        }
+        for key, pattern in patterns.items():
+            matches = re.findall(pattern, raw, re.IGNORECASE)
+            if matches:
+                if key in ("name_servers", "status"):
+                    data[key] = [m.strip().lower() for m in matches]
+                else:
+                    data[key] = matches[0].strip()
+        return data
+
+    def _enumerate_subdomains(self, domain: str) -> List[str]:
+        """Enumerate subdomains using crt.sh certificate transparency."""
+        try:
+            resp = self.http.get(
+                f"https://crt.sh/?q=%.{domain}&output=json",
+                headers={"Accept": "application/json"},
+            )
+            if resp and resp.status_code == 200:
+                certs = resp.json()
+                subdomains = set()
+                for cert in certs:
+                    name = cert.get("name_value", "")
+                    for entry in name.split("\n"):
+                        entry = entry.strip().lower()
+                        if entry.endswith(domain) and entry != domain:
+                            # Skip wildcards
+                            if not entry.startswith("*"):
+                                subdomains.add(entry)
+                return sorted(subdomains)
+        except Exception as e:
+            logger.debug(f"Subdomain enumeration failed for {domain}: {e}")
+        return []

eirescope/modules/ip_module.py

@@ -0,0 +1,218 @@
+"""IP Address OSINT Module — WHOIS, geolocation, DNS reverse lookup."""
+import re
+import socket
+import subprocess
+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.ip")
+
+
+class IPModule(BaseOSINTModule):
+    """IP address reconnaissance — WHOIS, GeoIP, reverse DNS."""
+
+    name = "IP Address Recon"
+    description = "Analyze IP: WHOIS lookup, geolocation, reverse DNS, ISP detection, abuse checks"
+    supported_entity_types = [EntityType.IP_ADDRESS]
+    requires_api_key = False
+    icon = "globe"
+
+    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 IP address reconnaissance."""
+        ip = entity.value.strip()
+        logger.info(f"Analyzing IP address: {ip}")
+        found_entities = []
+
+        # 1. GeoIP Lookup (free API)
+        geo_data = self._geoip_lookup(ip)
+        if geo_data:
+            entity.metadata["geolocation"] = geo_data
+            if geo_data.get("country"):
+                geo_entity = Entity(
+                    entity_type=EntityType.GEO_LOCATION,
+                    value=f"{geo_data.get('city', 'Unknown')}, {geo_data.get('country', 'Unknown')}",
+                    source_module=self.name,
+                    confidence=0.8,
+                    metadata=geo_data,
+                )
+                added = investigation.add_entity(geo_entity)
+                investigation.add_relationship(
+                    source_id=entity.id,
+                    target_id=added.id,
+                    rel_type="ip_located_in",
+                    confidence=0.8,
+                )
+                found_entities.append(added)
+
+            # ISP / Organization
+            if geo_data.get("isp") or geo_data.get("org"):
+                entity.metadata["isp"] = geo_data.get("isp", "")
+                entity.metadata["organization"] = geo_data.get("org", "")
+
+        # 2. Reverse DNS
+        rdns = self._reverse_dns(ip)
+        if rdns:
+            entity.metadata["reverse_dns"] = rdns
+            # Extract domain from reverse DNS
+            domain = self._extract_domain(rdns)
+            if domain:
+                domain_entity = Entity(
+                    entity_type=EntityType.DOMAIN,
+                    value=domain,
+                    source_module=self.name,
+                    confidence=0.7,
+                    metadata={"derived_from_ip": ip, "reverse_dns": rdns},
+                )
+                added = investigation.add_entity(domain_entity)
+                investigation.add_relationship(
+                    source_id=entity.id,
+                    target_id=added.id,
+                    rel_type="ip_resolves_to_domain",
+                    confidence=0.7,
+                )
+                found_entities.append(added)
+
+        # 3. WHOIS
+        whois_data = self._whois_lookup(ip)
+        if whois_data:
+            entity.metadata["whois"] = whois_data
+            whois_entity = Entity(
+                entity_type=EntityType.WHOIS_INFO,
+                value=f"WHOIS: {ip}",
+                source_module=self.name,
+                confidence=0.9,
+                metadata=whois_data,
+            )
+            added = investigation.add_entity(whois_entity)
+            investigation.add_relationship(
+                source_id=entity.id,
+                target_id=added.id,
+                rel_type="has_whois_record",
+                confidence=0.9,
+            )
+            found_entities.append(added)
+
+        # 4. Classify IP type
+        entity.metadata["ip_type"] = self._classify_ip(ip, geo_data)
+
+        logger.info(f"IP analysis complete: {len(found_entities)} entities discovered")
+        return found_entities
+
+    def _geoip_lookup(self, ip: str) -> Optional[Dict]:
+        """GeoIP lookup using free ip-api.com service."""
+        try:
+            resp = self.http.get(
+                f"http://ip-api.com/json/{ip}",
+                headers={"Accept": "application/json"},
+            )
+            if resp and resp.status_code == 200:
+                data = resp.json()
+                if data.get("status") == "success":
+                    return {
+                        "ip": ip,
+                        "country": data.get("country", ""),
+                        "country_code": data.get("countryCode", ""),
+                        "region": data.get("regionName", ""),
+                        "city": data.get("city", ""),
+                        "zip": data.get("zip", ""),
+                        "lat": data.get("lat"),
+                        "lon": data.get("lon"),
+                        "timezone": data.get("timezone", ""),
+                        "isp": data.get("isp", ""),
+                        "org": data.get("org", ""),
+                        "as": data.get("as", ""),
+                        "is_mobile": data.get("mobile", False),
+                        "is_proxy": data.get("proxy", False),
+                        "is_hosting": data.get("hosting", False),
+                    }
+        except Exception as e:
+            logger.debug(f"GeoIP lookup failed for {ip}: {e}")
+        return None
+
+    def _reverse_dns(self, ip: str) -> Optional[str]:
+        """Perform reverse DNS lookup."""
+        try:
+            hostname, _, _ = socket.gethostbyaddr(ip)
+            return hostname
+        except (socket.herror, socket.gaierror, OSError):
+            pass
+        # Fallback: use dig
+        try:
+            result = subprocess.run(
+                ["dig", "+short", "-x", ip],
+                capture_output=True, text=True, timeout=10,
+            )
+            if result.returncode == 0 and result.stdout.strip():
+                return result.stdout.strip().rstrip(".")
+        except Exception:
+            pass
+        return None
+
+    def _whois_lookup(self, ip: str) -> Optional[Dict]:
+        """WHOIS lookup using system whois command."""
+        try:
+            result = subprocess.run(
+                ["whois", ip],
+                capture_output=True, text=True, timeout=15,
+            )
+            if result.returncode == 0 and result.stdout:
+                return self._parse_whois(result.stdout)
+        except FileNotFoundError:
+            logger.debug("whois command not found")
+        except Exception as e:
+            logger.debug(f"WHOIS lookup failed for {ip}: {e}")
+        return None
+
+    def _parse_whois(self, raw: str) -> Dict:
+        """Parse raw WHOIS output into structured data."""
+        data = {"raw": raw[:2000]}  # Keep truncated raw
+        patterns = {
+            "netname": r"(?:NetName|netname):\s*(.+)",
+            "org_name": r"(?:OrgName|org-name|organisation):\s*(.+)",
+            "country": r"(?:Country|country):\s*(\S+)",
+            "address": r"(?:Address|address):\s*(.+)",
+            "cidr": r"(?:CIDR|inetnum):\s*(.+)",
+            "abuse_email": r"(?:OrgAbuseEmail|abuse-mailbox):\s*(\S+)",
+            "created": r"(?:RegDate|created):\s*(.+)",
+            "updated": r"(?:Updated|last-modified):\s*(.+)",
+        }
+        for key, pattern in patterns.items():
+            match = re.search(pattern, raw, re.IGNORECASE)
+            if match:
+                data[key] = match.group(1).strip()
+        return data
+
+    def _extract_domain(self, hostname: str) -> Optional[str]:
+        """Extract registrable domain from hostname."""
+        parts = hostname.rstrip(".").split(".")
+        if len(parts) >= 2:
+            return ".".join(parts[-2:])
+        return None
+
+    def _classify_ip(self, ip: str, geo_data: Optional[Dict] = None) -> str:
+        """Classify IP as residential, hosting, VPN, etc."""
+        if geo_data:
+            if geo_data.get("is_proxy"):
+                return "proxy/VPN"
+            if geo_data.get("is_hosting"):
+                return "hosting/datacenter"
+            if geo_data.get("is_mobile"):
+                return "mobile"
+
+        # Check private ranges
+        if ip.startswith(("10.", "172.16.", "172.17.", "172.18.", "172.19.",
+                          "172.20.", "172.21.", "172.22.", "172.23.", "172.24.",
+                          "172.25.", "172.26.", "172.27.", "172.28.", "172.29.",
+                          "172.30.", "172.31.", "192.168.")):
+            return "private"
+        if ip.startswith("127."):
+            return "loopback"
+
+        return "residential/unknown"

eirescope/modules/social_module.py

@@ -0,0 +1,184 @@
+"""Social Media OSINT Module — Profile discovery across platforms."""
+import logging
+from typing import List
+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.social")
+
+# Social platforms that allow email-based search or public profile lookup
+SOCIAL_SEARCH_URLS = {
+    "GitHub": "https://api.github.com/search/users?q={}",
+    "Gravatar": "https://en.gravatar.com/{}.json",
+}
+
+
+class SocialMediaModule(BaseOSINTModule):
+    """Cross-platform social media profile discovery."""
+
+    name = "Social Media Discovery"
+    description = "Find social media profiles linked to emails, usernames, or phone numbers"
+    supported_entity_types = [EntityType.EMAIL, EntityType.USERNAME]
+    requires_api_key = False
+    icon = "users"
+
+    def __init__(self, config=None):
+        super().__init__(config)
+        self.http = OSINTHTTPClient(timeout=8, max_retries=2, rate_limit=0.3)
+
+    def execute(self, entity: Entity, investigation: Investigation) -> List[Entity]:
+        """Discover social profiles from entity."""
+        logger.info(f"Social media discovery for: {entity.value} ({entity.entity_type.value})")
+        found_entities = []
+
+        if entity.entity_type == EntityType.USERNAME:
+            found_entities.extend(self._search_github_user(entity, investigation))
+
+        if entity.entity_type == EntityType.EMAIL:
+            found_entities.extend(self._search_by_email(entity, investigation))
+
+        logger.info(f"Social discovery complete: {len(found_entities)} profiles found")
+        return found_entities
+
+    def _search_github_user(self, entity: Entity, investigation: Investigation) -> List[Entity]:
+        """Search GitHub for user profile details."""
+        found = []
+        username = entity.value
+        try:
+            resp = self.http.get(f"https://api.github.com/users/{username}")
+            if resp and resp.status_code == 200:
+                data = resp.json()
+                profile_entity = Entity(
+                    entity_type=EntityType.SOCIAL_PROFILE,
+                    value=data.get("html_url", f"https://github.com/{username}"),
+                    source_module=self.name,
+                    confidence=0.95,
+                    metadata={
+                        "platform": "GitHub",
+                        "username": data.get("login"),
+                        "name": data.get("name"),
+                        "bio": data.get("bio"),
+                        "company": data.get("company"),
+                        "location": data.get("location"),
+                        "blog": data.get("blog"),
+                        "public_repos": data.get("public_repos"),
+                        "followers": data.get("followers"),
+                        "following": data.get("following"),
+                        "created_at": data.get("created_at"),
+                        "avatar_url": data.get("avatar_url"),
+                    },
+                )
+                added = investigation.add_entity(profile_entity)
+                investigation.add_relationship(
+                    source_id=entity.id,
+                    target_id=added.id,
+                    rel_type="has_github_profile",
+                    confidence=0.95,
+                )
+                found.append(added)
+
+                # Extract email from GitHub profile if public
+                if data.get("email"):
+                    email_entity = Entity(
+                        entity_type=EntityType.EMAIL,
+                        value=data["email"],
+                        source_module=self.name,
+                        confidence=0.9,
+                        metadata={"source": "github_profile"},
+                    )
+                    added_email = investigation.add_entity(email_entity)
+                    investigation.add_relationship(
+                        source_id=added.id,
+                        target_id=added_email.id,
+                        rel_type="profile_has_email",
+                        confidence=0.9,
+                    )
+                    found.append(added_email)
+
+                # Extract blog/website
+                if data.get("blog"):
+                    url_entity = Entity(
+                        entity_type=EntityType.URL,
+                        value=data["blog"],
+                        source_module=self.name,
+                        confidence=0.9,
+                        metadata={"source": "github_profile"},
+                    )
+                    added_url = investigation.add_entity(url_entity)
+                    investigation.add_relationship(
+                        source_id=added.id,
+                        target_id=added_url.id,
+                        rel_type="profile_links_to",
+                        confidence=0.9,
+                    )
+                    found.append(added_url)
+
+        except Exception as e:
+            logger.debug(f"GitHub search failed: {e}")
+        return found
+
+    def _search_by_email(self, entity: Entity, investigation: Investigation) -> List[Entity]:
+        """Search for profiles linked to an email address."""
+        found = []
+        email = entity.value
+
+        # Check Gravatar
+        import hashlib
+        email_hash = hashlib.md5(email.lower().encode()).hexdigest()
+        try:
+            resp = self.http.get(f"https://en.gravatar.com/{email_hash}.json")
+            if resp and resp.status_code == 200:
+                data = resp.json()
+                if "entry" in data and data["entry"]:
+                    entry = data["entry"][0]
+                    profile_entity = Entity(
+                        entity_type=EntityType.SOCIAL_PROFILE,
+                        value=entry.get("profileUrl", f"https://gravatar.com/{email_hash}"),
+                        source_module=self.name,
+                        confidence=0.9,
+                        metadata={
+                            "platform": "Gravatar",
+                            "display_name": entry.get("displayName"),
+                            "about": entry.get("aboutMe"),
+                            "location": entry.get("currentLocation"),
+                            "accounts": [
+                                {"platform": a.get("shortname"), "url": a.get("url")}
+                                for a in entry.get("accounts", [])
+                            ],
+                        },
+                    )
+                    added = investigation.add_entity(profile_entity)
+                    investigation.add_relationship(
+                        source_id=entity.id,
+                        target_id=added.id,
+                        rel_type="email_has_gravatar",
+                        confidence=0.9,
+                    )
+                    found.append(added)
+
+                    # Extract linked accounts from Gravatar
+                    for account in entry.get("accounts", []):
+                        if account.get("url"):
+                            acc_entity = Entity(
+                                entity_type=EntityType.SOCIAL_PROFILE,
+                                value=account["url"],
+                                source_module=self.name,
+                                confidence=0.85,
+                                metadata={
+                                    "platform": account.get("shortname", "unknown"),
+                                    "source": "gravatar_linked",
+                                },
+                            )
+                            added_acc = investigation.add_entity(acc_entity)
+                            investigation.add_relationship(
+                                source_id=added.id,
+                                target_id=added_acc.id,
+                                rel_type="gravatar_links_to",
+                                confidence=0.85,
+                            )
+                            found.append(added_acc)
+        except Exception as e:
+            logger.debug(f"Gravatar search failed: {e}")
+
+        return found