diff --git a/README.md b/README.md index 758ba12d..0c936684 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,7 @@ What PRISM does that those generally do not: results stream into a browser while | Censys | Host services, ASN, certificate → subdomain discovery | Censys | | VirusTotal | Domain/IP reputation, malware detections | VirusTotal | | AbuseIPDB | IP abuse confidence score | AbuseIPDB | -| Dark Web Checker | .onion mirrors via Ahmia + DarkSearch | none | +| Dark Web Checker | .onion mirrors via Ahmia | none | | Infostealer Exposure | Machines infected by stealers carrying the target's credentials (opt-in) | Hudson Rock | | Domain Exposure | Yearly exposure trend, malware families, affected services (opt-in) | Lunar | | Website Analyzer | Tech stack, emails, social links, metadata | none | @@ -367,7 +367,7 @@ prism/ │ ├── shodan_lookup.py # Shodan host intelligence │ ├── censys_lookup.py # Censys host + certificate search │ ├── wayback.py # Wayback Machine snapshots + sensitive URLs -│ ├── onion_checker.py # .onion mirror checker (Ahmia + DarkSearch) +│ ├── onion_checker.py # .onion mirror checker (Ahmia) │ ├── darkweb_search.py # Dark-web mentions search │ ├── blackbird.py # Username search (async, 50+ platforms) │ ├── maigret_wrapper.py # Deep username search (3000+ sites) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index fe6bc48c..ff74658f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -28,7 +28,7 @@ and how to extend it with new modules. │ ▼ 3rd-party APIs (Shodan, Censys, VT, AbuseIPDB, - Ahmia, DarkSearch, crt.sh, Wayback, HIBP, …) + Ahmia, crt.sh, Wayback, HIBP, …) ``` --- diff --git a/frontend/src/components/views/ScanResults.tsx b/frontend/src/components/views/ScanResults.tsx index 353ef46f..9b3e0ad7 100644 --- a/frontend/src/components/views/ScanResults.tsx +++ b/frontend/src/components/views/ScanResults.tsx @@ -1370,7 +1370,7 @@ export function ScanResults({ scan, onHome }: Props) { {tab === 'darkweb' && r.onion && ( refreshModule('onion')} refreshing={isRefreshing('onion')}>
- Sources: Ahmia ({r.onion.sources?.ahmia ?? 0}) · DarkSearch ({r.onion.sources?.darksearch ?? 0}) + Sources: Ahmia ({r.onion.sources?.ahmia ?? 0})
{r.onion.results?.map((item, i) => ( diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index 19103c20..1c1e5c6b 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -351,7 +351,7 @@ export interface OnionData { target?: string; total_found?: number; results?: { source?: string; url: string; title?: string | null; description?: string | null }[]; - sources?: { ahmia?: number; darksearch?: number }; + sources?: { ahmia?: number }; } export interface UrlScanResult { diff --git a/modules/onion_checker.py b/modules/onion_checker.py index dc5a9786..c327cfe1 100644 --- a/modules/onion_checker.py +++ b/modules/onion_checker.py @@ -2,16 +2,20 @@ from config import USER_AGENT import re -from typing import Any, Dict, List, Set +from typing import Any, Dict, List, Optional, Set, Tuple import requests from modules import get_proxies +from modules.module_status import ERROR, OK, annotate _ONION_RE = re.compile(r"https?://[a-z2-7]{16,56}\.onion[/\w\-\.]*", re.IGNORECASE) _TAG_RE = re.compile(r"<[^>]+>") _WS_RE = re.compile(r"\s+") +_HIDDEN_INPUT_RE = re.compile(r"]+type=[\"']hidden[\"'][^>]*>", re.IGNORECASE) +_NAME_RE = re.compile(r"name=[\"']([^\"']+)[\"']", re.IGNORECASE) +_VALUE_RE = re.compile(r"value=[\"']([^\"']+)[\"']", re.IGNORECASE) class OnionChecker: @@ -39,18 +43,55 @@ def _relevant(self, item: Dict[str, Any], tokens: Set[str]) -> bool: ).lower() return any(tok in hay for tok in tokens) - def _search_ahmia(self, query: str) -> List[Dict[str, Any]]: + def _extract_hidden_token(self, html: str) -> Optional[Tuple[str, str]]: + match = _HIDDEN_INPUT_RE.search(html or "") + if not match: + return None + tag = match.group(0) + name_m = _NAME_RE.search(tag) + val_m = _VALUE_RE.search(tag) + if name_m and val_m: + return name_m.group(1), val_m.group(1) + return None + + def _search_ahmia( + self, query: str, session: Optional[requests.Session] = None + ) -> Tuple[List[Dict[str, Any]], Optional[str]]: + """Search Ahmia for onion mirrors using a token from the home page. + + Returns (results, error_reason). + """ + s = session or requests.Session() + headers = {"User-Agent": USER_AGENT} + proxies = get_proxies() try: - r = requests.get( + r_home = s.get( + "https://ahmia.fi/", + timeout=self.timeout, + proxies=proxies, + headers=headers, + ) + if r_home.status_code != 200: + return [], f"Ahmia home page HTTP {r_home.status_code}" + + token_pair = self._extract_hidden_token(r_home.text) + if not token_pair: + return [], "Ahmia verification token not found on home page" + + token_name, token_value = token_pair + params = {"q": query, token_name: token_value} + + r_search = s.get( "https://ahmia.fi/search/", - params={"q": query}, + params=params, timeout=self.timeout, - proxies=get_proxies(), - headers={"User-Agent": USER_AGENT}, + proxies=proxies, + headers=headers, ) - if r.status_code != 200: - return [] - html = r.text or "" + if r_search.status_code != 200: + return [], f"Ahmia search HTTP {r_search.status_code}" + + html = r_search.text or "" out: List[Dict[str, Any]] = [] seen: Set[str] = set() for block in re.split(r']*class="[^"]*result', html)[1:]: @@ -63,50 +104,35 @@ def _search_ahmia(self, query: str) -> List[Dict[str, Any]]: seen.add(url) text = _WS_RE.sub(" ", _TAG_RE.sub(" ", block)).strip() out.append({"source": "ahmia", "url": url, "snippet": text[:400] or None}) - return out[:25] - except Exception: - return [] - - def _search_darksearch(self, query: str) -> List[Dict[str, Any]]: - try: - r = requests.get( - "https://darksearch.io/api/search", - params={"query": query, "page": 1}, - timeout=self.timeout, - proxies=get_proxies(), - headers={"User-Agent": USER_AGENT}, - ) - if r.status_code != 200: - return [] - data = r.json() - items = data.get("data") or [] - out: List[Dict[str, Any]] = [] - for it in items[:25]: - link = it.get("link") or "" - if ".onion" not in link: - continue - out.append({ - "source": "darksearch", - "url": link, - "title": (it.get("title") or "").strip()[:200] or None, - "description": (it.get("description") or "").strip()[:300] or None, - }) - return out - except Exception: - return [] + return out[:25], None + except Exception as e: + return [], f"Ahmia request failed: {e}" def check(self, target: str) -> Dict[str, Any]: target = (target or "").strip() if not target: - return {"target": target, "error": "empty target"} + res: Dict[str, Any] = {"target": target, "total_found": 0, "results": [], "sources": {"ahmia": 0}} + return annotate(res, ERROR, "empty target") tokens = self._tokens(target) - ahmia = self._search_ahmia(target) - darksearch = self._search_darksearch(target) + session = requests.Session() + ahmia, error = self._search_ahmia(target, session=session) + + base_res: Dict[str, Any] = { + "target": target, + "total_found": 0, + "results": [], + "sources": { + "ahmia": len(ahmia), + }, + } + + if error: + return annotate(base_res, ERROR, error) seen: Set[str] = set() merged: List[Dict[str, Any]] = [] - for item in ahmia + darksearch: + for item in ahmia: url = item.get("url", "") if not url or url in seen: continue @@ -116,13 +142,6 @@ def check(self, target: str) -> Dict[str, Any]: item.pop("snippet", None) merged.append(item) - return { - "target": target, - "total_found": len(merged), - "results": merged, - "sources": { - "ahmia": len(ahmia), - "darksearch": len(darksearch), - }, - "error": None, - } + base_res["total_found"] = len(merged) + base_res["results"] = merged + return annotate(base_res, OK) diff --git a/tests/test_v2_1_modules.py b/tests/test_v2_1_modules.py index fcd8fdd2..4209d8fa 100644 --- a/tests/test_v2_1_modules.py +++ b/tests/test_v2_1_modules.py @@ -9,45 +9,75 @@ class TestOnionChecker: def test_returns_error_on_empty_target(self): from modules.onion_checker import OnionChecker + from modules.module_status import classify, ERROR result = OnionChecker().check("") + assert classify(result) == ERROR assert result["error"] == "empty target" def test_aggregates_unique_relevant_results(self, monkeypatch): from modules.onion_checker import OnionChecker + from modules.module_status import classify, OK oc = OnionChecker() monkeypatch.setattr( oc, "_search_ahmia", - lambda q: [{"source": "ahmia", "url": "http://abc.onion", "snippet": "example.com leak dump"}], - ) - monkeypatch.setattr( - oc, "_search_darksearch", - lambda q: [ - {"source": "darksearch", "url": "http://abc.onion", "title": "example mirror"}, - {"source": "darksearch", "url": "http://xyz.onion", "title": "example dump"}, - {"source": "darksearch", "url": "http://unrelated.onion", "title": "random market"}, - ], + lambda q, session=None: ([ + {"source": "ahmia", "url": "http://abc.onion", "snippet": "example.com leak dump"}, + {"source": "ahmia", "url": "http://xyz.onion", "snippet": "example.com credentials"}, + {"source": "ahmia", "url": "http://unrelated.onion", "snippet": "random market"}, + ], None), ) result = oc.check("example.com") + assert classify(result) == OK assert result["error"] is None assert result["total_found"] == 2 urls = {r["url"] for r in result["results"]} assert urls == {"http://abc.onion", "http://xyz.onion"} assert "http://unrelated.onion" not in urls - assert result["sources"] == {"ahmia": 1, "darksearch": 3} + assert result["sources"] == {"ahmia": 3} + + def test_ahmia_token_extraction_and_search(self, monkeypatch): + import requests + from modules.onion_checker import OnionChecker + from modules.module_status import classify, OK - def test_ahmia_network_failure_handled(self, monkeypatch): + class MockResp: + def __init__(self, text, status_code=200): + self.text = text + self.status_code = status_code + + def fake_get(self, url, params=None, *a, **k): + if "search" in url: + assert params is not None + assert params.get("q") == "bitcoin" + assert params.get("85f590") == "986a59" + html = '
  • dump
    • ' + return MockResp(html) + # home page with rotating hidden token + return MockResp('
    ') + + monkeypatch.setattr(requests.Session, "get", fake_get) + oc = OnionChecker(timeout=5) + result = oc.check("bitcoin") + assert classify(result) == OK + assert result["total_found"] == 1 + assert result["results"][0]["url"] == "http://bitcoindump56charsxyz.onion/info" + + def test_ahmia_failure_reports_error_status(self, monkeypatch): import requests from modules.onion_checker import OnionChecker + from modules.module_status import classify, ERROR def raise_network(*a, **k): raise requests.exceptions.ConnectionError("boom") - monkeypatch.setattr(requests, "get", raise_network) + monkeypatch.setattr(requests.Session, "get", raise_network) result = OnionChecker(timeout=1).check("example.com") - - assert result["error"] is None + + assert classify(result) == ERROR + assert result["error"] is not None + assert "Ahmia request failed" in result["error"] assert result["total_found"] == 0