from __future__ import annotations import json import logging import os from dataclasses import dataclass, field from typing import Dict, Iterable, List, Optional, Sequence, Set, Tuple from urllib.parse import urlparse import requests from ..localization import t from ..localization.helpers import create_translated_finding from ..model import Finding, ScanRequest LOGGER = logging.getLogger("web_sentinel.checks.osint") CRT_SH_ENDPOINT = "https://crt.sh/" HIBP_ENDPOINT = "https://haveibeenpwned.com/api/v3/breachedaccount/{email}" CRT_RESULT_LIMIT = 200 SUBDOMAIN_ALERT_THRESHOLD = 10 @dataclass class EmailBreachOutcome: exposures: List[Dict[str, object]] = field(default_factory=list) skipped_reason: Optional[str] = None errors: List[str] = field(default_factory=list) @dataclass class OSINTCollector: request: ScanRequest session: requests.Session = field(default_factory=requests.Session, init=False) def __post_init__(self) -> None: self.domain = self._normalize_domain(self.request.domain) self.session.headers.update({"User-Agent": self.request.user_agent}) def collect_certificate_transparency_subdomains(self) -> Set[str]: """ Fetch subdomains registered in Certificate Transparency logs via crt.sh. """ if not self.domain: return set() params = { "q": self.domain, "output": "json", } try: response = self.session.get( CRT_SH_ENDPOINT, params=params, timeout=max(self.request.timeout, 5.0), ) response.raise_for_status() except requests.RequestException as exc: LOGGER.info(t("messages.osint_crtsh_lookup_failed").format(domain=self.domain, error=exc)) return set() try: certificates: Sequence[Dict[str, object]] = response.json() except ValueError: LOGGER.info(t("messages.osint_crtsh_invalid_json").format(domain=self.domain)) return set() subdomains: Set[str] = set() for entry in certificates[:CRT_RESULT_LIMIT]: name_value = str(entry.get("name_value", "")).strip() if not name_value: continue for candidate in name_value.split("\n"): cleaned = candidate.strip().lower() if not cleaned: continue if cleaned.startswith("*."): cleaned = cleaned[2:] if cleaned == self.domain: continue if cleaned.endswith(self.domain): subdomains.add(cleaned) return subdomains def detect_web_technologies(self) -> Dict[str, str]: """ Fingerprint HTTP application stack via headers and HTML heuristics. """ technologies: Dict[str, str] = {} probe_url, response = self._fetch_primary_document() if not response or not probe_url: return technologies header_map = {key.lower(): value for key, value in response.headers.items()} server = header_map.get("server") if server: technologies["web_server"] = server x_powered_by = header_map.get("x-powered-by") if x_powered_by: technologies["framework"] = x_powered_by powered_by = header_map.get("x-generator") if powered_by: technologies["generator"] = powered_by content = response.text.lower() cms_signatures = { "WordPress": ("wp-content", "wp-includes", "wordpress"), "Drupal": ("drupal/settings.js", "sites/default/files"), "Joomla": ("joomla-script-options", "com_content"), "Magento": ("mage/cookies.js", "magento"), "Shopify": ("cdn.shopify.com", "x-shopify-stage"), } for cms, signatures in cms_signatures.items(): if any(signature in content for signature in signatures): technologies["cms"] = cms break js_frameworks = { "React": ("data-reactroot", "react-dom"), "Vue.js": ("__vue__", "vue.runtime"), "Angular": ("ng-version", "angular"), "Svelte": ("svelte", "data-svelte"), "jQuery": ("jquery", "jquery.min.js"), } for framework, signatures in js_frameworks.items(): if any(signature in content for signature in signatures): technologies["frontend_framework"] = framework break cdn_headers = ("cf-ray", "cf-cache-status", "x-amz-cf-id", "x-fastly-request-id") if any(header in header_map for header in cdn_headers): technologies["cdn"] = "Cloudflare" if "cf-ray" in header_map else "CDN detected" if response.history: technologies["redirect_chain"] = " -> ".join( r.url for r in (*response.history, response) if getattr(r, "url", None) ) technologies["probe_url"] = probe_url return technologies def check_email_breaches(self) -> EmailBreachOutcome: """ Query HaveIBeenPwned API (if configured) for likely public email aliases. """ emails = self._candidate_emails() if not emails: return EmailBreachOutcome() api_key = os.getenv("HIBP_API_KEY") if not api_key: return EmailBreachOutcome(skipped_reason="missing_api_key") exposures: List[Dict[str, object]] = [] errors: List[str] = [] for email in emails: try: response = self.session.get( HIBP_ENDPOINT.format(email=email), headers={ "hibp-api-key": api_key, "User-Agent": self.session.headers.get("User-Agent", "WebSentinel/1.0"), }, timeout=max(self.request.timeout, 6.0), params={"truncateResponse": "false"}, ) except requests.RequestException as exc: errors.append(f"{email}: network error {exc}") continue if response.status_code == 200: try: breaches = response.json() except ValueError: errors.append(f"{email}: invalid JSON response") continue exposures.append( { "email": email, "breaches": [ { "name": breach.get("Name"), "title": breach.get("Title"), "added_date": breach.get("AddedDate"), "pwn_count": breach.get("PwnCount"), } for breach in breaches ], } ) continue if response.status_code == 404: continue if response.status_code in (401, 403): return EmailBreachOutcome( exposures=exposures, skipped_reason=f"http_{response.status_code}", ) if response.status_code == 429: return EmailBreachOutcome( exposures=exposures, skipped_reason="rate_limited", ) errors.append(f"{email}: HTTP {response.status_code}") return EmailBreachOutcome(exposures=exposures, errors=errors) def _fetch_primary_document(self) -> Tuple[Optional[str], Optional[requests.Response]]: targets = [ f"https://{self.domain}", f"http://{self.domain}", ] for url in targets: try: response = self.session.get( url, timeout=max(self.request.timeout, 5.0), allow_redirects=True, ) return response.url, response except requests.RequestException as exc: LOGGER.debug("HTTP fingerprint probe failed for %s via %s: %s", self.domain, url, exc) continue return None, None def _candidate_emails(self) -> List[str]: if not self.domain: return [] base = self.domain if base.count(".") < 1: return [] prefixes = ("admin", "info", "contact", "support", "security") return [f"{prefix}@{base}" for prefix in prefixes] @staticmethod def _normalize_domain(raw: str) -> str: raw = raw.strip() parsed = urlparse(raw if raw.startswith("http") else f"//{raw}") netloc = parsed.netloc or parsed.path return netloc.lower().strip("/") def evaluate_osint_footprint(request: ScanRequest) -> Iterable[Finding]: collector = OSINTCollector(request) findings: List[Finding] = [] subdomains = collector.collect_certificate_transparency_subdomains() if len(subdomains) >= SUBDOMAIN_ALERT_THRESHOLD: preview = ", ".join(sorted(subdomains)[:10]) findings.append( create_translated_finding( check="osint", i18n_key="osint.subdomains.large_footprint", severity="medium", i18n_params={ "domain": collector.domain, "count": len(subdomains), "examples": preview, }, evidence=preview ) ) technologies = collector.detect_web_technologies() if technologies: findings.append( create_translated_finding( check="osint", i18n_key="osint.technologies.detected", severity="info", i18n_params={"technologies": json.dumps(technologies, ensure_ascii=False)}, evidence=json.dumps(technologies, indent=2, sort_keys=True) ) ) email_outcome = collector.check_email_breaches() if email_outcome.exposures: exposed_addresses = ", ".join(item["email"] for item in email_outcome.exposures) findings.append( create_translated_finding( check="osint", i18n_key="osint.email.exposed", severity="low", i18n_params={ "emails": exposed_addresses, }, evidence=json.dumps(email_outcome.exposures, indent=2, sort_keys=True) ) ) if email_outcome.skipped_reason: findings.append( create_translated_finding( check="osint", i18n_key="osint.email.lookup_skipped", severity="info", i18n_params={"reason": email_outcome.skipped_reason}, evidence=email_outcome.skipped_reason ) ) elif email_outcome.errors: findings.append( create_translated_finding( check="osint", i18n_key="osint.email.lookup_partial_errors", severity="low", i18n_params={"errors": "; ".join(email_outcome.errors)}, evidence="; ".join(email_outcome.errors) ) ) return findings def _describe_email_skip(reason: str) -> str: if reason == "missing_api_key": return "HaveIBeenPwned API key not configured; email breach lookup skipped." if reason.startswith("http_"): status = reason.split("_", 1)[1] return f"HaveIBeenPwned API responded with HTTP {status}; email breach lookup skipped." if reason == "rate_limited": return "HaveIBeenPwned API rate limit reached; email breach lookup skipped." return f"Email breach lookup skipped: {reason}"