fix(onion_checker): handle Ahmia verification token, drop dead DarkSearch, annotate module_status (#535)

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)

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, …)
 ```
 
 ---

frontend/src/components/views/ScanResults.tsx

@@ -1370,7 +1370,7 @@ export function ScanResults({ scan, onHome }: Props) {
         {tab === 'darkweb' && r.onion && (
           <Card title={`Dark Web Mirrors - ${r.onion.total_found} found`} onRefresh={() => refreshModule('onion')} refreshing={isRefreshing('onion')}>
             <div className="text-[11px] text-text-3 mb-3">
-              Sources: Ahmia ({r.onion.sources?.ahmia ?? 0}) · DarkSearch ({r.onion.sources?.darksearch ?? 0})
+              Sources: Ahmia ({r.onion.sources?.ahmia ?? 0})
             </div>
             <div className="space-y-2">
               {r.onion.results?.map((item, i) => (

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 {

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"<input[^>]+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'<li[^>]*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)

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 = '<li class="result"><a href="http://bitcoindump56charsxyz.onion/info">dump</a></li>'
+                return MockResp(html)
+            # home page with rotating hidden token
+            return MockResp('<html><form action="/search/"><input type="hidden" name="85f590" value="986a59"></form></html>')
+
+        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