From a04f6791140a2fd53fb5ca5ce9503c23a5eee982 Mon Sep 17 00:00:00 2001 From: osintph Date: Sat, 4 Jul 2026 15:11:55 +0800 Subject: [PATCH 1/8] quick-scan: add storage models and Storage methods for quick_scan_sessions and quick_scan_findings Co-Authored-By: Claude Fable 5 --- src/darkweb_scanner/storage.py | 146 +++++++++++++++++++++++++++++++++ 1 file changed, 146 insertions(+) diff --git a/src/darkweb_scanner/storage.py b/src/darkweb_scanner/storage.py index 9eb0f24..17399ef 100644 --- a/src/darkweb_scanner/storage.py +++ b/src/darkweb_scanner/storage.py @@ -291,6 +291,48 @@ class CustomIntel(Base): ) +# ── Quick Scan Models ─────────────────────────────────────────────────────────── + +class QuickScanSession(Base): + __tablename__ = "quick_scan_sessions" + + id = Column(Integer, primary_key=True, autoincrement=True) + user_id = Column(Integer, nullable=False) + target_value = Column(String(512), nullable=False) + target_type = Column(String(50), nullable=False) # email | domain | url | company_name + normalized_variants = Column(Text, default="[]") # JSON list of strings searched + sources_used = Column(Text, default="[]") # JSON list of source names + status = Column(String(50), default="pending") # pending|running|completed|failed|cancelled + started_at = Column(DateTime, default=lambda: datetime.now(timezone.utc).replace(tzinfo=None)) + completed_at = Column(DateTime, nullable=True) + urls_visited = Column(Integer, default=0) + findings_count = Column(Integer, default=0) + error_message = Column(Text, nullable=True) + + __table_args__ = ( + Index("ix_quick_scan_sessions_target", "target_value"), + Index("ix_quick_scan_sessions_user", "user_id"), + ) + + +class QuickScanFinding(Base): + __tablename__ = "quick_scan_findings" + + id = Column(Integer, primary_key=True, autoincrement=True) + session_id = Column(Integer, nullable=False) + source_name = Column(String(200), nullable=False) + url = Column(Text, nullable=False) + matched_variant = Column(String(512), nullable=False) + context = Column(Text) + high_signal = Column(Boolean, default=False) + found_at = Column(DateTime, default=lambda: datetime.now(timezone.utc).replace(tzinfo=None)) + + __table_args__ = ( + Index("ix_quick_scan_findings_session", "session_id"), + Index("ix_quick_scan_findings_source", "source_name"), + ) + + class Storage: def __init__(self, database_url: Optional[str] = None): self.database_url = database_url or os.getenv( @@ -1321,3 +1363,107 @@ def get_last_hit_date(self, keywords: list[str]) -> str | None: .scalar() ) return result.isoformat() if result else None + + # --- Quick Scan Sessions & Findings --- + + def create_quick_scan_session( + self, + user_id: int, + target_value: str, + target_type: str, + normalized_variants: list[str], + sources_used: list[str], + ) -> int: + with self.get_session() as session: + record = QuickScanSession( + user_id=user_id, + target_value=target_value, + target_type=target_type, + normalized_variants=json.dumps(normalized_variants), + sources_used=json.dumps(sources_used), + status="pending", + ) + session.add(record) + session.commit() + session.refresh(record) + return record.id + + def update_quick_scan_session(self, session_id: int, **kwargs) -> None: + with self.get_session() as session: + record = session.get(QuickScanSession, session_id) + if not record: + return + for key, value in kwargs.items(): + if key in ("normalized_variants", "sources_used") and isinstance(value, (list, dict)): + value = json.dumps(value) + setattr(record, key, value) + session.commit() + + def add_quick_scan_finding( + self, + session_id: int, + source_name: str, + url: str, + matched_variant: str, + context: str, + high_signal: bool, + ) -> int: + with self.get_session() as session: + record = QuickScanFinding( + session_id=session_id, + source_name=source_name, + url=url, + matched_variant=matched_variant, + context=context, + high_signal=high_signal, + ) + session.add(record) + session.commit() + session.refresh(record) + return record.id + + def get_quick_scan_session(self, session_id: int) -> Optional["QuickScanSession"]: + with self.get_session() as session: + record = session.get(QuickScanSession, session_id) + if record is not None: + session.expunge(record) + return record + + def list_quick_scan_sessions(self, user_id: int, limit: int = 50) -> list["QuickScanSession"]: + with self.get_session() as session: + records = ( + session.query(QuickScanSession) + .filter(QuickScanSession.user_id == user_id) + .order_by(QuickScanSession.id.desc()) + .limit(limit) + .all() + ) + for record in records: + session.expunge(record) + return records + + def list_quick_scan_findings( + self, session_id: int, high_signal_only: bool = False + ) -> list["QuickScanFinding"]: + with self.get_session() as session: + query = session.query(QuickScanFinding).filter( + QuickScanFinding.session_id == session_id + ) + if high_signal_only: + query = query.filter(QuickScanFinding.high_signal.is_(True)) + records = query.order_by(QuickScanFinding.id.asc()).all() + for record in records: + session.expunge(record) + return records + + def has_active_quick_scan(self, user_id: int) -> bool: + with self.get_session() as session: + record = ( + session.query(QuickScanSession.id) + .filter( + QuickScanSession.user_id == user_id, + QuickScanSession.status.in_(("pending", "running")), + ) + .first() + ) + return record is not None From 76a2584d450c23cb17d4063e651d3073ca460259 Mon Sep 17 00:00:00 2001 From: osintph Date: Sat, 4 Jul 2026 15:17:24 +0800 Subject: [PATCH 2/8] quick-scan: add source config module and normalization logic in quick_scan.py Co-Authored-By: Claude Fable 5 --- src/darkweb_scanner/quick_scan.py | 164 ++++++++++++++++++++ src/darkweb_scanner/quick_scan_sources.py | 180 ++++++++++++++++++++++ 2 files changed, 344 insertions(+) create mode 100644 src/darkweb_scanner/quick_scan.py create mode 100644 src/darkweb_scanner/quick_scan_sources.py diff --git a/src/darkweb_scanner/quick_scan.py b/src/darkweb_scanner/quick_scan.py new file mode 100644 index 0000000..24d0bff --- /dev/null +++ b/src/darkweb_scanner/quick_scan.py @@ -0,0 +1,164 @@ +""" +Quick Scan — ad-hoc target investigation, independent of the project/IOC pipeline. + +This module holds: + 1. Target auto-detection and normalization (pure functions). + 2. "High signal" classification and context-window extraction (pure functions). + 3. The async orchestrator ``run_quick_scan`` (added in a later change). + +Detection and normalization are deliberately pure and side-effect free so they can +be unit-tested exhaustively without any network or database. +""" + +import logging +import re +from urllib.parse import urlparse + +logger = logging.getLogger(__name__) + +# Target types +TYPE_EMAIL = "email" +TYPE_DOMAIN = "domain" +TYPE_URL = "url" +TYPE_COMPANY = "company_name" +VALID_TARGET_TYPES = (TYPE_EMAIL, TYPE_DOMAIN, TYPE_URL, TYPE_COMPANY) + +# Keywords that, if present in a match's context window, flag it as high signal. +HIGH_SIGNAL_KEYWORDS = ( + "leaked", "leak", "dump", "combo", "credentials", "database", + "breach", "hacked", "for sale", "selling", "buy", +) + +# Context window: characters kept on each side of a match. +CONTEXT_SIDE = 100 + +_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") +# label.tld, one or more labels, ascii TLD >= 2 chars. No scheme, no path. +_DOMAIN_RE = re.compile( + r"^(?=.{1,253}$)([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,}$", + re.IGNORECASE, +) + + +def detect_target_type(raw: str) -> str: + """Auto-detect the target type from the raw input string. + + Order (per spec): + 1. contains '@' and matches an email regex -> email + 2. starts with http:// or https:// -> url + 3. matches a domain (label.tld) regex -> domain + 4. otherwise -> company_name + """ + value = (raw or "").strip() + if "@" in value and _EMAIL_RE.match(value): + return TYPE_EMAIL + lowered = value.lower() + if lowered.startswith("http://") or lowered.startswith("https://"): + return TYPE_URL + if _DOMAIN_RE.match(value): + return TYPE_DOMAIN + return TYPE_COMPANY + + +def _dedupe(items) -> list: + """Order-preserving de-duplication, dropping empties.""" + seen = set() + out = [] + for item in items: + if item and item not in seen: + seen.add(item) + out.append(item) + return out + + +def _apex_of(hostname: str) -> str: + """Naive apex: the last two dot-labels. Does not handle multi-part TLDs + (e.g. co.uk); acceptable for this iteration's variant expansion.""" + labels = hostname.split(".") + if len(labels) >= 2: + return ".".join(labels[-2:]) + return hostname + + +def normalize_url(raw: str) -> str: + """Normalize a URL to 'hostname[/path]', dropping scheme, query and fragment.""" + parsed = urlparse(raw.strip()) + host = (parsed.hostname or "").lower() + path = parsed.path or "" + if path == "/": + path = "" + return f"{host}{path}" + + +def normalize_variants(target_value: str, target_type: str) -> list: + """Return the ordered, de-duplicated list of strings to actually search for.""" + value = (target_value or "").strip() + + if target_type == TYPE_EMAIL: + local, _, domain = value.partition("@") + return _dedupe([value, local, domain]) + + if target_type == TYPE_URL: + normalized = normalize_url(value) + host = normalized.split("/", 1)[0] + return _dedupe([normalized, host]) + + if target_type == TYPE_DOMAIN: + host = value.lower() + variants = [host] + if not host.startswith("www."): + variants.append("www." + host) + # If this is a subdomain, also search the apex. + apex = _apex_of(host) + if apex != host: + variants.append(apex) + return _dedupe(variants) + + # company_name: exact string only, no fuzzy expansion at this stage. + return _dedupe([value]) + + +def is_high_signal(context: str) -> bool: + """True if the context window contains any high-signal keyword (case-insensitive).""" + if not context: + return False + lowered = context.lower() + return any(keyword in lowered for keyword in HIGH_SIGNAL_KEYWORDS) + + +def extract_context(text: str, start: int, end: int) -> str: + """Extract the 200-char window centered on a match: 100 before + match + 100 after.""" + lo = max(0, start - CONTEXT_SIDE) + hi = min(len(text), end + CONTEXT_SIDE) + return text[lo:hi] + + +def find_matches(text: str, variants: list) -> list: + """Find the first occurrence of each variant in ``text`` (case-insensitive). + + Returns a list of dicts: {variant, context, high_signal}. One entry per variant + that appears at least once, so a single page produces at most len(variants) + findings regardless of repetition. + + TODO: LLM relevance scoring would plug in here — re-rank / filter these matches + with a model call before they become findings. + """ + if not text: + return [] + haystack = text.lower() + results = [] + for variant in variants: + if not variant: + continue + idx = haystack.find(variant.lower()) + if idx == -1: + continue + context = extract_context(text, idx, idx + len(variant)) + results.append( + { + "variant": variant, + "context": context, + "high_signal": is_high_signal(context), + } + ) + return results diff --git a/src/darkweb_scanner/quick_scan_sources.py b/src/darkweb_scanner/quick_scan_sources.py new file mode 100644 index 0000000..ee25569 --- /dev/null +++ b/src/darkweb_scanner/quick_scan_sources.py @@ -0,0 +1,180 @@ +""" +Quick Scan source registry. + +Plain-Python, typed source definitions for the ad-hoc "Quick Scan" feature. +Kept as code (not YAML/JSON) on purpose: adding, removing, or re-pointing a +source is a reviewable diff. + +Transport model: + - ``tor`` : fetched through the existing Tor SOCKS5 client (.onion hosts). + - ``clearnet`` : fetched through ``dashboard.http_client.safe_fetch`` — HTTPS-only, + host-allowlisted, IP-blocklisted. Any clearnet source host MUST be + present in ALLOWED_EXTERNAL_HOSTS or the fetch is refused. + +A source exposes one or more ``url_templates``. A template containing the literal +``{query}`` placeholder is a search endpoint (the URL-encoded target is substituted +in). A template with no placeholder is a fixed seed URL fetched as-is (used by leak +sites that have no search box). +""" + +from dataclasses import dataclass +from urllib.parse import quote + +from .ransomware_data import RANSOMWARE_ONION_SEEDS + +# Transport identifiers +TRANSPORT_TOR = "tor" +TRANSPORT_CLEARNET = "clearnet" + +# Source kinds +KIND_SEARCH_ENGINE = "search_engine" +KIND_LEAK_SITE = "leak_site" +KIND_PASTE = "paste" +KIND_FORUM = "forum" + + +@dataclass(frozen=True) +class QuickScanSource: + """A single queryable Quick Scan source.""" + + name: str # stable identifier used in sources_used and the API + label: str # human-facing label for the UI checklist + kind: str # one of the KIND_* constants + transport: str # TRANSPORT_TOR | TRANSPORT_CLEARNET + url_templates: tuple # tuple[str, ...]; "{query}" marks a search endpoint + enabled: bool = True # disabled sources are never queried + + @property + def search_capable(self) -> bool: + return any("{query}" in t for t in self.url_templates) + + def build_urls(self, query: str) -> list: + """Return the concrete URLs to fetch for a given target query. + + Search templates get the URL-encoded query substituted; fixed seed + templates are returned unchanged. A source with no templates yields an + empty list and is therefore skipped cleanly by the orchestrator. + """ + encoded = quote(query, safe="") + urls = [] + for template in self.url_templates: + if "{query}" in template: + urls.append(template.replace("{query}", encoded)) + else: + urls.append(template) + return urls + + +# ── Dark web search engines (onion, via Tor) ──────────────────────────────────── + +_SEARCH_ENGINES = [ + QuickScanSource( + name="ahmia", + label="Ahmia", + kind=KIND_SEARCH_ENGINE, + transport=TRANSPORT_TOR, + url_templates=( + "http://juhanurmihxlp77nkq76byazcldy2hlmovfu2epvl5ankdibsot4csyd.onion/search/?q={query}", + ), + ), + QuickScanSource( + name="torch", + label="Torch", + kind=KIND_SEARCH_ENGINE, + transport=TRANSPORT_TOR, + url_templates=( + "http://xmh57jrknzkhv6y3ls3ubitzfqnkrwxhopf5aygthi7d6rplyvk3noyd.onion/search?q={query}", + ), + ), + QuickScanSource( + name="haystak", + label="Haystak", + kind=KIND_SEARCH_ENGINE, + transport=TRANSPORT_TOR, + url_templates=( + "http://haystak5njsmn2hqkewecpaxetahtwhsbsa64jom2k22z5afxhnpxfid.onion/?q={query}", + ), + ), +] + + +# ── Ransomware leak sites (onion, via Tor) ─────────────────────────────────────── +# Reuse the curated seed list as-is. These have no search box, so each seed page is +# fetched and crawled for the target rather than queried. + +_RANSOMWARE_LEAKS = [ + QuickScanSource( + name="ransomware_leaks", + label="Ransomware Leak Sites", + kind=KIND_LEAK_SITE, + transport=TRANSPORT_TOR, + url_templates=tuple(RANSOMWARE_ONION_SEEDS), + ), +] + + +# ── Paste sites (clearnet, via safe_fetch) ─────────────────────────────────────── +# Neither rentry.co nor dpaste.org exposes a documented public *search* endpoint, so +# both ship disabled (empty url_templates -> skipped cleanly). Populate a search +# template and flip enabled=True once an endpoint is confirmed. dpaste.org is kept in +# ALLOWED_EXTERNAL_HOSTS so enabling it is a one-line change, not a security review. + +_PASTE_SITES = [ + QuickScanSource( + name="dpaste", + label="dpaste.org", + kind=KIND_PASTE, + transport=TRANSPORT_CLEARNET, + url_templates=(), # no public search endpoint yet; add "https://dpaste.org/...{query}" + enabled=False, + ), + QuickScanSource( + name="rentry", + label="rentry.co", + kind=KIND_PASTE, + transport=TRANSPORT_CLEARNET, + url_templates=(), # no public search endpoint; skip cleanly + enabled=False, + ), +] + + +# ── Forum seed list (onion, curated) ───────────────────────────────────────────── +# Intentionally empty. Populate before enabling forum sources — forum onion addresses +# rotate frequently and stale hard-coded URLs are worse than none. + +_FORUMS: list = [ + # QuickScanSource(name="...", label="...", kind=KIND_FORUM, + # transport=TRANSPORT_TOR, url_templates=("http://...onion/search?q={query}",)), +] + + +# All registered sources, in display order. +ALL_SOURCES: tuple = tuple(_SEARCH_ENGINES + _RANSOMWARE_LEAKS + _PASTE_SITES + _FORUMS) + +_BY_NAME = {s.name: s for s in ALL_SOURCES} + + +def get_source(name: str): + """Return the source with the given name, or None.""" + return _BY_NAME.get(name) + + +def default_enabled_sources() -> list: + """Sources used when the caller does not specify an explicit subset.""" + return [s for s in ALL_SOURCES if s.enabled] + + +def resolve_sources(names) -> list: + """Resolve a list of requested source names to enabled QuickScanSource objects. + + Unknown or disabled names are dropped silently. ``None`` means "all enabled". + """ + if names is None: + return default_enabled_sources() + resolved = [] + for name in names: + source = _BY_NAME.get(name) + if source is not None and source.enabled: + resolved.append(source) + return resolved From dc72d03b79ce8e5aca4c9a08f3c6b95fa2a38b0d Mon Sep 17 00:00:00 2001 From: osintph Date: Sat, 4 Jul 2026 15:22:41 +0800 Subject: [PATCH 3/8] quick-scan: add orchestrator run_quick_scan with tor and safe_fetch integration Co-Authored-By: Claude Fable 5 --- src/darkweb_scanner/quick_scan.py | 230 +++++++++++++++++++++++++++++- 1 file changed, 229 insertions(+), 1 deletion(-) diff --git a/src/darkweb_scanner/quick_scan.py b/src/darkweb_scanner/quick_scan.py index 24d0bff..a245c1a 100644 --- a/src/darkweb_scanner/quick_scan.py +++ b/src/darkweb_scanner/quick_scan.py @@ -10,9 +10,18 @@ be unit-tested exhaustively without any network or database. """ +import asyncio +import json import logging +import os +import random import re -from urllib.parse import urlparse +import time +from datetime import datetime, timezone +from urllib.parse import urljoin, urlparse + +import aiohttp +from bs4 import BeautifulSoup logger = logging.getLogger(__name__) @@ -32,6 +41,13 @@ # Context window: characters kept on each side of a match. CONTEXT_SIDE = 100 +# Crawl behavior caps (module-level so tests can monkeypatch them). +MAX_DEPTH = 2 # hops followed from a source's initial fetch +MAX_URLS_PER_SCAN = 100 # hard cap on URLs attempted per scan +PER_SOURCE_TIMEOUT = 30 # seconds per individual fetch +TOTAL_TIMEOUT = 600 # 10-minute hard cap on the whole scan +MAX_CONCURRENT = int(os.getenv("QUICK_SCAN_MAX_CONCURRENT", "4")) + _EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") # label.tld, one or more labels, ascii TLD >= 2 chars. No scheme, no path. _DOMAIN_RE = re.compile( @@ -162,3 +178,215 @@ def find_matches(text: str, variants: list) -> list: } ) return results + + +# ── Orchestration ──────────────────────────────────────────────────────────────── + + +def _now() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def _is_onion(url: str) -> bool: + return (urlparse(url).hostname or "").endswith(".onion") + + +def _normalize_url(url: str) -> str: + return urlparse(url)._replace(fragment="").geturl().rstrip("/") + + +def _rate_limit_delay() -> float: + """Reuse the crawler's random-delay rate-limit approach. + + Reads the crawler's CRAWL_DELAY_* knobs so behavior matches the main crawler, + with QUICK_SCAN_DELAY_* overrides (tests set these to 0 for speed). + """ + lo = float(os.getenv("QUICK_SCAN_DELAY_MIN", os.getenv("CRAWL_DELAY_MIN", "2"))) + hi = float(os.getenv("QUICK_SCAN_DELAY_MAX", os.getenv("CRAWL_DELAY_MAX", "8"))) + if hi < lo: + hi = lo + return random.uniform(lo, hi) + + +def _extract_page_text(html: str) -> str: + soup = BeautifulSoup(html, "lxml") + for tag in soup(["script", "style", "noscript"]): + tag.decompose() + return soup.get_text(separator=" ", strip=True) + + +def _extract_links(html: str, base_url: str) -> list: + soup = BeautifulSoup(html, "lxml") + links = [] + for tag in soup.find_all("a", href=True): + href = tag["href"].strip() + if not href or href.startswith(("#", "mailto:", "javascript:")): + continue + absolute = urljoin(base_url, href) + parsed = urlparse(absolute) + if parsed.scheme in ("http", "https"): + links.append(_normalize_url(absolute)) + return links + + +async def _fetch_onion(url: str, tor_client) -> tuple: + session = await tor_client.get_session() + timeout = aiohttp.ClientTimeout(total=PER_SOURCE_TIMEOUT) + async with session.get(url, ssl=False, allow_redirects=True, timeout=timeout) as resp: + html = await resp.text(errors="replace") + return resp.status, html + + +async def _fetch_clearnet(url: str) -> tuple: + """Fetch a clearnet URL through safe_fetch (HTTPS-only, allowlisted, IP-blocked). + + Imported lazily and run in a thread since safe_fetch is synchronous. + """ + from .dashboard.http_client import safe_fetch + + def _do(): + result = safe_fetch(url, timeout=PER_SOURCE_TIMEOUT, allow_redirects=True) + body = result.get("body") or b"" + if isinstance(body, bytes): + body = body.decode("utf-8", "replace") + return result.get("status", 0), body + + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, _do) + + +async def _fetch(url: str, tor_client) -> tuple: + """Return (status_code, html). Onion via Tor, clearnet via safe_fetch.""" + if _is_onion(url): + return await _fetch_onion(url, tor_client) + return await _fetch_clearnet(url) + + +async def run_quick_scan(session_id: int, storage, tor_client) -> None: + """Run a quick scan end to end for a persisted session. + + Loads the session, marks it running, queries each configured source with the + target value, fetches result pages (and links up to MAX_DEPTH), records findings + whenever a normalized variant appears in page text, and finalizes the session. + + Per-URL errors are logged and skipped. A total-time overrun finalizes the session + as ``completed`` with a warning in error_message. Only failures that prevent the + scan from running at all mark it ``failed``. + """ + from .quick_scan_sources import resolve_sources + + session = storage.get_quick_scan_session(session_id) + if session is None: + logger.warning("Quick scan session %s not found; aborting", session_id) + return + + target_value = session.target_value + variants = json.loads(session.normalized_variants or "[]") + source_names = json.loads(session.sources_used or "[]") + sources = resolve_sources(source_names) + + storage.update_quick_scan_session(session_id, status="running", started_at=_now()) + + deadline = time.monotonic() + TOTAL_TIMEOUT + semaphore = asyncio.Semaphore(MAX_CONCURRENT) + visited: set = set() + counters = {"urls_visited": 0, "findings": 0} + timed_out = False + + async def _process(url: str, depth: int, source_name: str) -> list: + """Fetch one URL, record findings, return discovered links.""" + async with semaphore: + if time.monotonic() > deadline: + return [] + await asyncio.sleep(_rate_limit_delay()) + try: + status, html = await asyncio.wait_for( + _fetch(url, tor_client), timeout=PER_SOURCE_TIMEOUT + ) + except Exception as exc: # noqa: BLE001 — per-URL errors must not abort the scan + logger.info("Quick scan fetch failed for %s: %s", url, exc) + return [] + counters["urls_visited"] += 1 + text = _extract_page_text(html) + for match in find_matches(text, variants): + storage.add_quick_scan_finding( + session_id=session_id, + source_name=source_name, + url=url, + matched_variant=match["variant"], + context=match["context"], + high_signal=match["high_signal"], + ) + counters["findings"] += 1 + return _extract_links(html, url) + + # Seed the frontier with each enabled source's query URLs (depth 0). + queue: list = [] + for source in sources: + for url in source.build_urls(target_value): + norm = _normalize_url(url) + if norm in visited: + continue + if len(visited) >= MAX_URLS_PER_SCAN: + break + visited.add(norm) + queue.append((url, 0, source.name)) + + try: + while queue: + if time.monotonic() > deadline: + timed_out = True + break + batch = queue[:MAX_CONCURRENT] + queue = queue[MAX_CONCURRENT:] + results = await asyncio.gather( + *[_process(url, depth, name) for url, depth, name in batch], + return_exceptions=True, + ) + for (url, depth, name), result in zip(batch, results): + if isinstance(result, Exception): + logger.warning("Quick scan task error for %s: %s", url, result) + continue + if depth >= MAX_DEPTH: + continue + for link in result: + norm = _normalize_url(link) + if norm in visited: + continue + if len(visited) >= MAX_URLS_PER_SCAN: + break + visited.add(norm) + queue.append((link, depth + 1, name)) + except Exception as exc: # noqa: BLE001 — finalize as failed on unexpected error + logger.exception("Quick scan %s failed", session_id) + storage.update_quick_scan_session( + session_id, + status="failed", + completed_at=_now(), + urls_visited=counters["urls_visited"], + findings_count=counters["findings"], + error_message=f"Scan failed: {type(exc).__name__}", + ) + return + + warning = None + if timed_out: + warning = ( + f"Scan reached the {TOTAL_TIMEOUT}s total time cap before all sources " + "were exhausted; results may be partial." + ) + storage.update_quick_scan_session( + session_id, + status="completed", + completed_at=_now(), + urls_visited=counters["urls_visited"], + findings_count=counters["findings"], + error_message=warning, + ) + logger.info( + "Quick scan %s complete: %d URLs, %d findings%s", + session_id, + counters["urls_visited"], + counters["findings"], + " (timed out)" if timed_out else "", + ) From 4c4e998b131846d33d81f7f23621ab957eeb29dc Mon Sep 17 00:00:00 2001 From: osintph Date: Sat, 4 Jul 2026 15:24:04 +0800 Subject: [PATCH 4/8] quick-scan: add /api/quick-scan/* routes Adds start/status/sessions/findings endpoints plus the /quick-scan page route, all @require_login. Ownership checks return 403; a DB-backed active-scan guard returns 409. Adds dpaste.org to ALLOWED_EXTERNAL_HOSTS for the clearnet paste path. Co-Authored-By: Claude Fable 5 --- .../dashboard/dashboard_routes.py | 169 ++++++++++++++++++ src/darkweb_scanner/dashboard/http_client.py | 4 + 2 files changed, 173 insertions(+) diff --git a/src/darkweb_scanner/dashboard/dashboard_routes.py b/src/darkweb_scanner/dashboard/dashboard_routes.py index 9c2e664..5167f75 100644 --- a/src/darkweb_scanner/dashboard/dashboard_routes.py +++ b/src/darkweb_scanner/dashboard/dashboard_routes.py @@ -3600,6 +3600,175 @@ def whiteintel_search(): return jsonify({"ok": False, "error": f"HTTP {status}"}), 200 +# ── Quick Scan ───────────────────────────────────────────────────────────────── + + +def _iso(dt): + return dt.isoformat() if dt else None + + +def _quick_scan_session_dict(sess) -> dict: + return { + "id": sess.id, + "target_value": sess.target_value, + "target_type": sess.target_type, + "normalized_variants": json.loads(sess.normalized_variants or "[]"), + "sources_used": json.loads(sess.sources_used or "[]"), + "status": sess.status, + "started_at": _iso(sess.started_at), + "completed_at": _iso(sess.completed_at), + "urls_visited": sess.urls_visited, + "findings_count": sess.findings_count, + "error_message": sess.error_message, + } + + +@dashboard_bp.route("/quick-scan") +@require_login +def quick_scan_page(): + from ..quick_scan_sources import ALL_SOURCES + + sources = [ + {"name": s.name, "label": s.label, "kind": s.kind, + "transport": s.transport, "enabled": s.enabled} + for s in ALL_SOURCES + ] + return render_template( + "quick_scan.html", + username=session.get("username"), + sources=sources, + ) + + +@dashboard_bp.route("/api/quick-scan/start", methods=["POST"]) +@require_login +def api_quick_scan_start(): + import asyncio + import threading + + from ..quick_scan import ( + VALID_TARGET_TYPES, + detect_target_type, + normalize_variants, + run_quick_scan, + ) + from ..quick_scan_sources import resolve_sources + from ..tor_client import create_tor_client + + storage = get_storage() + user_id = session["user_id"] + + body = request.get_json(silent=True) or {} + target = (body.get("target") or "").strip() + if not target: + return jsonify({"error": "target is required"}), 400 + + target_type = body.get("target_type") + if target_type in (None, "", "auto"): + target_type = detect_target_type(target) + elif target_type not in VALID_TARGET_TYPES: + return jsonify({"error": f"invalid target_type {target_type!r}"}), 400 + + variants = normalize_variants(target, target_type) + + requested_sources = body.get("sources") + sources = resolve_sources(requested_sources) + if not sources: + return jsonify({"error": "no enabled sources selected"}), 400 + source_names = [s.name for s in sources] + + # Single active quick scan per user (DB-backed so it holds across workers). + if storage.has_active_quick_scan(user_id): + return jsonify({"error": "A quick scan is already running"}), 409 + + session_id = storage.create_quick_scan_session( + user_id=user_id, + target_value=target, + target_type=target_type, + normalized_variants=variants, + sources_used=source_names, + ) + + def run(): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + tor_client = create_tor_client() + try: + loop.run_until_complete(run_quick_scan(session_id, storage, tor_client)) + except Exception as exc: # noqa: BLE001 + logger.exception("Quick scan thread crashed for session %s", session_id) + try: + storage.update_quick_scan_session( + session_id, + status="failed", + error_message=f"Scan crashed: {type(exc).__name__}", + ) + except Exception: + pass + finally: + try: + loop.run_until_complete(tor_client.close()) + except Exception: + pass + loop.close() + + threading.Thread(target=run, daemon=True, name="quick_scan_thread").start() + + return jsonify({"session_id": session_id}), 201 + + +@dashboard_bp.route("/api/quick-scan/status/") +@require_login +def api_quick_scan_status(session_id): + storage = get_storage() + sess = storage.get_quick_scan_session(session_id) + if sess is None: + return jsonify({"error": "not found"}), 404 + if sess.user_id != session["user_id"]: + return jsonify({"error": "forbidden"}), 403 + return jsonify({ + "status": sess.status, + "started_at": _iso(sess.started_at), + "completed_at": _iso(sess.completed_at), + "urls_visited": sess.urls_visited, + "findings_count": sess.findings_count, + "error_message": sess.error_message, + }) + + +@dashboard_bp.route("/api/quick-scan/sessions") +@require_login +def api_quick_scan_sessions(): + storage = get_storage() + sessions = storage.list_quick_scan_sessions(session["user_id"], limit=50) + return jsonify([_quick_scan_session_dict(s) for s in sessions]) + + +@dashboard_bp.route("/api/quick-scan/findings/") +@require_login +def api_quick_scan_findings(session_id): + storage = get_storage() + sess = storage.get_quick_scan_session(session_id) + if sess is None: + return jsonify({"error": "not found"}), 404 + if sess.user_id != session["user_id"]: + return jsonify({"error": "forbidden"}), 403 + high_signal_only = request.args.get("high_signal_only") == "1" + findings = storage.list_quick_scan_findings(session_id, high_signal_only=high_signal_only) + return jsonify([ + { + "id": f.id, + "source_name": f.source_name, + "url": f.url, + "matched_variant": f.matched_variant, + "context": f.context, + "high_signal": f.high_signal, + "found_at": _iso(f.found_at), + } + for f in findings + ]) + + # ── Health ───────────────────────────────────────────────────────────────────── diff --git a/src/darkweb_scanner/dashboard/http_client.py b/src/darkweb_scanner/dashboard/http_client.py index 0fdc63f..0bff400 100644 --- a/src/darkweb_scanner/dashboard/http_client.py +++ b/src/darkweb_scanner/dashboard/http_client.py @@ -75,6 +75,10 @@ "whiteintel.io", # Mail infrastructure "api.mailgun.net", + # Quick Scan clearnet paste sources (see quick_scan_sources.py). The dpaste + # source ships disabled pending a search endpoint; the host is pre-allowlisted + # so enabling it is a one-line change rather than a security review. + "dpaste.org", }) # Supplementary networks not fully covered by ipaddress stdlib classifiers on all From b7a086cc2c98253582a63f5f669f79743d1f6b5d Mon Sep 17 00:00:00 2001 From: osintph Date: Sat, 4 Jul 2026 15:26:47 +0800 Subject: [PATCH 5/8] quick-scan: add Quick Scan tab UI (template, JS, sidebar entry) Co-Authored-By: Claude Fable 5 --- .../dashboard/templates/index.html | 1 + .../dashboard/templates/quick_scan.html | 357 ++++++++++++++++++ 2 files changed, 358 insertions(+) create mode 100644 src/darkweb_scanner/dashboard/templates/quick_scan.html diff --git a/src/darkweb_scanner/dashboard/templates/index.html b/src/darkweb_scanner/dashboard/templates/index.html index 5d553ea..95e0de3 100644 --- a/src/darkweb_scanner/dashboard/templates/index.html +++ b/src/darkweb_scanner/dashboard/templates/index.html @@ -466,6 +466,7 @@

Threat Intelligence Platform