research/websearch.py

"""Live web / dark-web retrieval layer for the tiny researcher (client hands).

The 16M model is the analyst brain and cannot browse. This module is the
client-side retrieval tool: it searches the open web (Google News RSS,
Internet Archive, Wikipedia - all key-free), optionally reaches .onion
services through a local Tor SOCKS5 proxy, extracts documents into plain
text, and writes them into the local library so TinyIndex can surface them
for the model to verify, cross-check, and guide the user down rabbit holes.

Guardrails (research/OSINT only):
  * read-only, no identity, no credentials, no execution, size caps.
  * http(s) and .onion only; other schemes (file:// ftp:// etc.) refused.
  * .onion requests need a local Tor SOCKS proxy; if it is not running the
    caller gets a clear, actionable message (never a silent empty result).
"""

from __future__ import annotations

import html as _html
import json
import re
import socket
import ssl
import time
from pathlib import Path

import requests
import urllib.parse

SOCKS_HOST = "127.0.0.1"
SOCKS_PORT = 9050
UA = "FSI-forensic-research/0.1 (research OSINT only) Mozilla/5.0"
MAX_BODY = 1_500_000       # raw bytes cap per source
MAX_CHARS = 120_000        # extracted text cap stored per doc
TIMEOUT = (12, 25)         # requests (connect, read)


def tor_status():
    """Probe the local Tor SOCKS proxy. Returns (ok: bool, msg: str)."""
    try:
        with socket.create_connection((SOCKS_HOST, SOCKS_PORT), timeout=2):
            return True, "Tor SOCKS is up on {0}:{1}".format(SOCKS_HOST, SOCKS_PORT)
    except OSError:
        return False, (
            "Tor not reachable on {0}:{1}. Start a local Tor daemon "
            "(tor, Tor Browser, or Orbot) and retry.".format(SOCKS_HOST, SOCKS_PORT)
        )


def _socks5_connect(host, port, timeout=5):
    """Open a TCP socket to (host, port) through the local Tor SOCKS5 proxy."""
    s = socket.create_connection((SOCKS_HOST, SOCKS_PORT), timeout=timeout)
    try:
        s.settimeout(timeout)
        s.sendall(b"\x05\x01\x00")            # SOCKS5, 1 method: no-auth
        rep = s.recv(2)
        if rep != b"\x05\x00":
            raise ConnectionError("Tor proxy requires auth (not supported)")
        raw_host = host.encode("ascii", errors="ignore")
        if len(raw_host) > 255:
            raise ValueError("hostname too long for SOCKS5")
        s.sendall(b"\x05\x01\x00\x03" + bytes([len(raw_host)]) + raw_host
                  + port.to_bytes(2, "big"))
        head = s.recv(512)                    # VER REP RSV ATYP ADDR BND.PORT
        if len(head) < 2 or head[1] != 0x00:
            raise ConnectionError("SOCKS5 CONNECT refused for {0}:{1}".format(host, port))
        return s
    except Exception:
        s.close()
        raise


def _http_over_socks(host, port, https, path, timeout=25):
    """Send one HTTP/1.1 GET over a Tor TCP socket (TLS-wrapped if https)."""
    s = _socks5_connect(host, port, min(timeout, 10))
    try:
        if https:
            ctx = ssl.create_default_context()
            s = ctx.wrap_socket(s, server_hostname=host)
        s.settimeout(timeout)
        host_hdr = host if (https and port == 443) else "{0}:{1}".format(host, port)
        req = ("GET {0} HTTP/1.1\r\nHost: {1}\r\nUser-Agent: {2}\r\n"
               "Accept: text/html\r\nConnection: close\r\n\r\n").format(
                   path, host_hdr, UA)
        s.sendall(req.encode())
        buf = b""
        while len(buf) < MAX_BODY:
            try:
                chunk = s.recv(65536)
            except (socket.timeout, OSError):
                break
            if not chunk:
                break
            buf += chunk
        return buf
    finally:
        try:
            s.close()
        except Exception:
            pass


def _split_http(buf):
    idx = buf.find(b"\r\n\r\n")
    if idx < 0:
        return buf, b""
    return buf[:idx], buf[idx + 4:]


def _extract_text(raw):
    if isinstance(raw, (bytes, bytearray)):
        txt = bytes(raw).decode("utf-8", errors="replace")
    else:
        txt = str(raw)
    txt = re.sub(r"(?is)<(script|style|head|header|footer|nav)[^>]*>.*?</\1>", " ", txt)
    txt = re.sub(r"(?i)<br\s*/?>[\r\n]*", "\\n", txt)
    txt = re.sub(r"(?s)<[^>]+>", " ", txt)
    txt = _html.unescape(txt)
    txt = re.sub(r"[ \\t]+", " ", txt)
    txt = re.sub(r"\n\s*\n+", "\n\n", txt)
    return txt.strip()


def _first_title(html_str):
    m = re.search(r"(?is)<title[^>]*>(.*?)</title>", html_str)
    if not m:
        return "(untitled)"
    return _extract_text(m.group(1))[:160] or "(untitled)"


def fetch(url, tor=False, timeout=25):
    """Fetch one URL (clearnet or .onion) into a dict with plain text."""
    u = urllib.parse.urlparse(url)
    if u.scheme not in ("http", "https"):
        raise ValueError("refusing non-http(s) target: {0}".format(u.scheme))
    is_onion = (u.hostname or "").endswith(".onion") or tor
    if is_onion:
        ok, msg = tor_status()
        if not ok:
            raise RuntimeError(msg)
        port = u.port or (443 if u.scheme == "https" else 80)
        path = (u.path or "/") + (("?" + u.query) if u.query else "")
        raw = _http_over_socks(u.hostname, port, u.scheme == "https", path, timeout)
        head, body = _split_http(raw)
        content = _extract_text(body or raw)
        status = re.search(br"HTTP/1\.[01] (\d{3})", head)
        title = _first_title(raw.decode("utf-8", "replace"))
        return {"url": url, "title": title, "content": content,
                "source": "onion",
                "status": (status.group(1).decode() if status else "?")}
    r = requests.get(url, headers={"User-Agent": UA}, timeout=TIMEOUT,
                     allow_redirects=True)
    r.raise_for_status()
    return {"url": r.url, "title": _first_title(r.text) or "(untitled)",
            "content": _extract_text(r.content),
            "source": "clearnet", "status": str(r.status_code)}


def search_news(query, limit=10):
    url = ("https://news.google.com/rss/search?q=" + urllib.parse.quote(query)
           + "&hl=en-US&gl=US&ceid=US:en")
    r = requests.get(url, headers={"User-Agent": UA}, timeout=TIMEOUT)
    r.raise_for_status()
    out = []
    for item in re.findall(r"(?is)<item>(.*?)</item>", r.text)[:limit]:
        t = re.search(r"(?is)<title>(.*?)</title>", item)
        link = re.search(r"(?is)<link>(.*?)</link>", item)
        desc = re.search(r"(?is)<description>(.*?)</description>", item)
        pub = re.search(r"(?is)<pubDate>(.*?)</pubDate>", item)
        if not link:
            continue
        out.append({
            "title": (_html.unescape(t.group(1)) if t else "(news)").strip(),
            "url": link.group(1).strip(),
            "snippet": (_html.unescape(desc.group(1)).strip() if desc else ""),
            "source": "google-news",
            "date": (pub.group(1).strip() if pub else ""),
        })
    return out


def search_archive(query, limit=5):
    params = {"q": query, "fl[]": ["identifier", "title"],
              "rows": limit, "output": "json"}
    r = requests.get("https://archive.org/advancedsearch.php", params=params,
                     headers={"User-Agent": UA}, timeout=TIMEOUT)
    r.raise_for_status()
    docs = r.json().get("response", {}).get("docs", [])
    out = []
    for d in docs:
        if not d.get("identifier"):
            continue
        out.append({"title": (d.get("title") or d["identifier"]).strip(),
                    "url": "https://archive.org/details/" + d["identifier"],
                    "snippet": "Internet Archive item", "source": "archive-org",
                    "date": ""})
    return out[:limit]


def search_wiki(query, limit=4):
    p = {"action": "query", "list": "search", "srsearch": query,
         "format": "json", "srlimit": limit}
    r = requests.get("https://en.wikipedia.org/w/api.php", params=p,
                     headers={"User-Agent": UA}, timeout=TIMEOUT)
    r.raise_for_status()
    res = r.json().get("query", {}).get("search", [])
    out = []
    for d in res:
        out.append({
            "title": d["title"],
            "url": "https://en.wikipedia.org/wiki/" + urllib.parse.quote(
                d["title"].replace(" ", "_")),
            "snippet": re.sub(r"<.*?>", "", d.get("snippet", "")),
            "source": "wikipedia", "date": ""})
    return out


def search_web(query, limit=10):
    """Search clearnet across news + archive + wiki; dedup by URL."""
    combined = []
    for fn in (search_news, search_archive, search_wiki):
        try:
            combined += fn(query, limit=max(1, limit // 2 + 1))
        except Exception:
            continue
    seen, out = set(), []
    for r in combined:
        if r["url"] in seen:
            continue
        seen.add(r["url"])
        out.append(r)
        if len(out) >= limit:
            break
    return out


def save_doc(library_dir, slug, title, text):
    """Write one pulled document into the library as a .txt file. Returns Path."""
    if not text:
        raise ValueError("no text to save")
    safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", slug)[:60].strip("_") or "doc"
    p = Path(library_dir) / ("pull_{0}_{1}.txt".format(int(time.time()), safe))
    p.write_text("TITLE: {0}\nSOURCE: {1}\n\n{2}".format(title, slug, text[:MAX_CHARS]),
                 encoding="utf-8")
    return p


def pull(query, n=3, tor=False, library_dir="data/library"):
    """Search, fetch the top-n docs, save each into the library. Returns dict."""
    results = search_web(query, limit=max(n * 3, 6))
    saved, errors = [], []
    for r in results:
        if len(saved) >= n:
            break
        try:
            doc = fetch(r["url"], tor=tor)
            if not doc["content"]:
                errors.append({"url": r["url"], "err": "empty body"})
                continue
            p = save_doc(library_dir, r["url"].rsplit("/", 1)[-1], doc["title"],
                         doc["content"])
            saved.append({"url": r["url"], "title": doc["title"], "file": str(p)})
        except Exception as e:
            errors.append({"url": r["url"], "err": str(e)[:200]})
    return {"query": query, "saved": saved, "errors": errors,
            "tor": tor, "tor_status": tor_status()}


if __name__ == "__main__":
    import argparse
    ap = argparse.ArgumentParser(description="live retrieval for the tiny researcher")
    ap.add_argument("--search", help="search clearnet for a topic")
    ap.add_argument("--fetch", help="fetch one URL")
    ap.add_argument("--pull", help="search + fetch top docs into the library")
    ap.add_argument("--tor", action="store_true", help="route fetches via Tor SOCKS")
    ap.add_argument("--library", default="data/library")
    ap.add_argument("--n", type=int, default=3)
    args = ap.parse_args()

    if args.search:
        for r in search_web(args.search, limit=8):
            print("[{0}] {1}\n    {2}\n    {3}".format(
                r["source"], r["title"], r["url"], r["snippet"][:120]))
    elif args.pull:
        res = pull(args.pull, library_dir=args.library)
        print(json.dumps(res, indent=2, ensure_ascii=False))
    elif args.fetch:
        print(json.dumps(fetch(args.fetch, tor=args.tor), indent=2, ensure_ascii=False))
    else:
        ap.print_help()