2c6281d041
Expand manhuang collect/UI (tabs, markets, guishi/quote probes), add configs and launchd helpers, ops handbooks, and scripts for channel validation and price/threat enrichment. Keys stay in keys.env.example only.
846 lines
32 KiB
Python
Executable file
846 lines
32 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""鬼市核验探针：Tor/SOCKS + 清网入口健康 + onion 白名单探活 + 品牌撞库 feed 抽样。
|
||
|
||
写出：~/.cache/manhuang/guishi/status.json
|
||
读：
|
||
~/.config/manhuang/onion-markets.json（可选）
|
||
~/.config/manhuang/shops.json 或 ~/data/manhuang/shops.json（本地店表）
|
||
~/.config/manhuang/keys.env（可选 BRAND_DOMAINS / RSSHUB_BASE）
|
||
TOR_SOCKS 默认 127.0.0.1:9050（备选 9150）
|
||
|
||
不写密钥；onion 仅探活，不爬店内商品；feed 只做自有品牌子串匹配。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import socket
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
UA = "joy-manhuang-guishi-probe/1.0 (+health; scout-tor; no-secrets)"
|
||
|
||
DEFAULT_BRANDS = [
|
||
"feibisi.com",
|
||
"weixiaoduo.com",
|
||
"wenpai.org",
|
||
"cravatar.com",
|
||
"feicode.com",
|
||
"linuxjoy",
|
||
]
|
||
|
||
|
||
def _data_dir() -> Path:
|
||
"""scout 约定 data/manhuang；Mac 回退 ~/.cache。可用 MANHUANG_DATA 覆盖。"""
|
||
env = (os.environ.get("MANHUANG_DATA") or "").strip()
|
||
if env:
|
||
return Path(env).expanduser() / "guishi"
|
||
home = Path.home()
|
||
scout = home / "data" / "manhuang" / "guishi"
|
||
if (home / "data" / "manhuang").is_dir():
|
||
return scout
|
||
return home / ".cache" / "manhuang" / "guishi"
|
||
|
||
|
||
OUT_DIR = _data_dir()
|
||
OUT = OUT_DIR / "status.json"
|
||
ONION_CFG = Path.home() / ".config" / "manhuang" / "onion-markets.json"
|
||
# scout 侧可选
|
||
ONION_CFG_ALT = Path.home() / "data" / "manhuang" / "onion-markets.json"
|
||
ONION_EX = Path(__file__).resolve().parents[2] / "configs" / "manhuang" / "onion-markets.example.json"
|
||
KEYS = Path.home() / ".config" / "manhuang" / "keys.env"
|
||
SHOPS_PATHS = (
|
||
Path.home() / ".config" / "manhuang" / "shops.json",
|
||
Path.home() / "data" / "manhuang" / "shops.json",
|
||
)
|
||
SHOPS_EX = Path(__file__).resolve().parents[2] / "configs" / "manhuang" / "shops.example.json"
|
||
|
||
# 清网可先探（不依赖 Tor）— 浅水入口 + 论坛健康 + paste
|
||
CLEARNET_CHECKS: list[dict[str, str]] = [
|
||
{"id": "ahmia", "url": "https://ahmia.fi/", "kind": "onion_search"},
|
||
{"id": "ahmia-onions", "url": "https://ahmia.fi/onions/", "kind": "onion_index"},
|
||
{"id": "darkfail", "url": "https://dark.fail/", "kind": "mirror_status"},
|
||
{"id": "onion-live", "url": "https://onion.live/", "kind": "onion_catalog"},
|
||
{"id": "torproject-check", "url": "https://check.torproject.org/", "kind": "tor_docs"},
|
||
{"id": "tor-bridges", "url": "https://bridges.torproject.org/", "kind": "tor_bridge"},
|
||
{"id": "onionland", "url": "https://onionlandsearchengine.com/", "kind": "onion_search"},
|
||
{"id": "urlscan", "url": "https://urlscan.io/", "kind": "surface"},
|
||
{"id": "threatfox", "url": "https://threatfox.abuse.ch/", "kind": "threat"},
|
||
{"id": "urlhaus", "url": "https://urlhaus.abuse.ch/", "kind": "threat"},
|
||
{"id": "malwarebazaar", "url": "https://bazaar.abuse.ch/", "kind": "threat"},
|
||
{"id": "feodo", "url": "https://feodotracker.abuse.ch/", "kind": "threat"},
|
||
{"id": "ransomware-live", "url": "https://www.ransomware.live/", "kind": "leak_watch"},
|
||
{"id": "ransomlook-home", "url": "https://www.ransomlook.io/", "kind": "leak_watch"},
|
||
{"id": "intelx-home", "url": "https://intelx.io/", "kind": "leak_watch"},
|
||
{"id": "5sim", "url": "https://5sim.net/", "kind": "sms_clear"},
|
||
{"id": "ggsel", "url": "https://ggsel.net/", "kind": "heishi_ref"},
|
||
{"id": "funpay", "url": "https://funpay.com/en/", "kind": "heishi_ref"},
|
||
{"id": "plati", "url": "https://plati.market/?lang=en-US", "kind": "heishi_ref"},
|
||
{"id": "linuxdo-search", "url": "https://linux.do/search?q=chatgpt", "kind": "grey_forum"},
|
||
{"id": "hostloc-home", "url": "https://hostloc.com/", "kind": "grey_forum"},
|
||
{"id": "nodeseek-home", "url": "https://www.nodeseek.com/", "kind": "grey_forum"},
|
||
{"id": "bhw-home", "url": "https://www.blackhatworld.com/", "kind": "grey_forum"},
|
||
{"id": "bhw-account-selling", "url": "https://www.blackhatworld.com/forums/account-selling-renting-services.310/", "kind": "grey_forum"},
|
||
{"id": "v2ex-home", "url": "https://www.v2ex.com/", "kind": "grey_forum"},
|
||
{"id": "pastebin", "url": "https://pastebin.com/", "kind": "paste"},
|
||
{"id": "dpaste", "url": "https://dpaste.org/", "kind": "paste"},
|
||
{"id": "rentry", "url": "https://rentry.co/", "kind": "paste"},
|
||
{"id": "leakix-home", "url": "https://leakix.net/", "kind": "surface"},
|
||
{"id": "phishtank", "url": "https://phishtank.org/", "kind": "phishing"},
|
||
{"id": "real-world-onion-gh", "url": "https://github.com/alecmuffett/real-world-onion-sites", "kind": "onion_catalog"},
|
||
{"id": "goofish", "url": "https://www.goofish.com/", "kind": "c2c_cn"},
|
||
]
|
||
|
||
# 公开 JSON / 文本 feed（无 key 或弱依赖）
|
||
JSON_CHECKS: list[dict[str, str]] = [
|
||
{
|
||
"id": "cisa-kev",
|
||
"url": "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json",
|
||
"kind": "threat_feed",
|
||
},
|
||
{
|
||
"id": "onionoo-summary",
|
||
"url": "https://onionoo.torproject.org/summary?limit=1",
|
||
"kind": "tor_metrics",
|
||
},
|
||
{
|
||
"id": "hibp-latest",
|
||
"url": "https://haveibeenpwned.com/api/v3/latestBreach",
|
||
"kind": "leak_breach",
|
||
},
|
||
{
|
||
"id": "5sim-guest-products",
|
||
"url": "https://5sim.net/v1/guest/products/any/any",
|
||
"kind": "sms_api",
|
||
},
|
||
{
|
||
"id": "urlscan-brand",
|
||
"url": "https://urlscan.io/api/v1/search/?q=domain:feibisi.com&size=5",
|
||
"kind": "brand_url_exposure",
|
||
},
|
||
{
|
||
"id": "urlscan-brand-wenpai",
|
||
"url": "https://urlscan.io/api/v1/search/?q=domain:wenpai.org&size=5",
|
||
"kind": "brand_url_exposure",
|
||
},
|
||
{
|
||
"id": "hibp-breaches",
|
||
"url": "https://haveibeenpwned.com/api/v3/breaches",
|
||
"kind": "leak_breach",
|
||
},
|
||
{
|
||
"id": "ransomware-api-groups",
|
||
"url": "https://api.ransomware.live/v2/groups",
|
||
"kind": "ransomware_api",
|
||
},
|
||
{
|
||
"id": "ransomware-api-recent",
|
||
"url": "https://api.ransomware.live/v2/recentvictims",
|
||
"kind": "ransomware_api",
|
||
},
|
||
{
|
||
"id": "ransomlook-posts",
|
||
"url": "https://www.ransomlook.io/api/posts",
|
||
"kind": "ransomware_api",
|
||
},
|
||
{
|
||
"id": "openphish-feed",
|
||
"url": "https://raw.githubusercontent.com/openphish/public_feed/refs/heads/main/feed.txt",
|
||
"kind": "phishing_feed",
|
||
},
|
||
{
|
||
"id": "phishing-db-today",
|
||
"url": "https://raw.githubusercontent.com/mitchellkrogza/Phishing.Database/master/phishing-links-ACTIVE-today.txt",
|
||
"kind": "phishing_feed",
|
||
},
|
||
{
|
||
"id": "maltrail-onion",
|
||
"url": "https://raw.githubusercontent.com/stamparm/maltrail/master/trails/static/suspicious/onion.txt",
|
||
"kind": "onion_ioc",
|
||
},
|
||
{
|
||
"id": "feodo-ipblocklist",
|
||
"url": "https://feodotracker.abuse.ch/downloads/ipblocklist.json",
|
||
"kind": "threat_feed",
|
||
},
|
||
{
|
||
"id": "rwos-commits",
|
||
"url": "https://api.github.com/repos/alecmuffett/real-world-onion-sites/commits?per_page=1",
|
||
"kind": "onion_catalog",
|
||
},
|
||
{
|
||
"id": "ransomwatch-posts",
|
||
"url": "https://ransomwhat.telemetry.ltd/posts",
|
||
"kind": "ransomware_api",
|
||
},
|
||
]
|
||
|
||
|
||
def _now() -> str:
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
def _load_keys() -> None:
|
||
if not KEYS.is_file():
|
||
return
|
||
try:
|
||
for line in KEYS.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, _, v = line.partition("=")
|
||
k, v = k.strip(), v.strip().strip('"').strip("'")
|
||
if k and k not in os.environ:
|
||
os.environ[k] = v
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _tcp_open(host: str, port: int, timeout: float = 2.0) -> bool:
|
||
try:
|
||
with socket.create_connection((host, port), timeout=timeout):
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def _detect_socks() -> dict[str, Any]:
|
||
env = (os.environ.get("TOR_SOCKS") or "").strip()
|
||
candidates: list[tuple[str, int]] = []
|
||
if env and ":" in env:
|
||
h, _, p = env.rpartition(":")
|
||
try:
|
||
candidates.append((h or "127.0.0.1", int(p)))
|
||
except ValueError:
|
||
pass
|
||
candidates.extend([("127.0.0.1", 9050), ("127.0.0.1", 9150)])
|
||
for host, port in candidates:
|
||
if _tcp_open(host, port):
|
||
return {"ok": True, "host": host, "port": port, "proxy": f"socks5h://{host}:{port}"}
|
||
return {"ok": False, "host": "127.0.0.1", "port": 9050, "proxy": None, "error": "no_socks"}
|
||
|
||
|
||
def _fetch(
|
||
url: str,
|
||
*,
|
||
timeout: int = 12,
|
||
proxy: str | None = None,
|
||
direct: bool = True,
|
||
keep_body: bool = False,
|
||
body_limit: int = 120_000,
|
||
) -> dict[str, Any]:
|
||
"""优先 curl：清网 --noproxy '*'；Tor --socks5-hostname（不依赖 PySocks）。"""
|
||
import shutil
|
||
import subprocess
|
||
|
||
t0 = time.monotonic()
|
||
curl = shutil.which("curl")
|
||
if curl:
|
||
cmd = [
|
||
curl, "-sS", "-L", "--max-time", str(timeout),
|
||
"-A", UA, "-o", "-", "-w", "\n%{http_code}",
|
||
]
|
||
env = os.environ.copy()
|
||
if proxy and not direct:
|
||
# socks5h://127.0.0.1:9050
|
||
hp = proxy.split("://", 1)[-1]
|
||
h, _, p = hp.rpartition(":")
|
||
cmd.extend(["--socks5-hostname", f"{h}:{p}"])
|
||
# 走 Tor 时清掉系统 http_proxy，避免双代理
|
||
for k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "all_proxy"):
|
||
env.pop(k, None)
|
||
else:
|
||
# 清网：默认绕过坏 Clash（访问外网仍可能需代理，故先 noproxy 再 fallback）
|
||
cmd.extend(["--noproxy", "*"])
|
||
for k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "all_proxy"):
|
||
env.pop(k, None)
|
||
cmd.append(url)
|
||
try:
|
||
r = subprocess.run(cmd, capture_output=True, timeout=timeout + 5, env=env)
|
||
raw = r.stdout.decode("utf-8", "replace")
|
||
lines = raw.rsplit("\n", 1)
|
||
body = lines[0] if len(lines) == 2 else raw
|
||
try:
|
||
code = int(lines[-1].strip() or "0") if len(lines) == 2 else 0
|
||
except ValueError:
|
||
code = 0
|
||
body = raw
|
||
ms = int((time.monotonic() - t0) * 1000)
|
||
if code == 0 and r.returncode != 0:
|
||
# noproxy 失败且是 clearnet：再试系统代理一次
|
||
if direct or not proxy:
|
||
env2 = os.environ.copy()
|
||
cmd2 = [
|
||
curl, "-sS", "-L", "--max-time", str(timeout),
|
||
"-A", UA, "-o", "-", "-w", "\n%{http_code}", url,
|
||
]
|
||
r2 = subprocess.run(cmd2, capture_output=True, timeout=timeout + 5, env=env2)
|
||
raw2 = r2.stdout.decode("utf-8", "replace")
|
||
parts = raw2.rsplit("\n", 1)
|
||
body = parts[0] if len(parts) == 2 else raw2
|
||
try:
|
||
code = int(parts[-1].strip() or "0") if len(parts) == 2 else 0
|
||
except ValueError:
|
||
code = 0
|
||
ms = int((time.monotonic() - t0) * 1000)
|
||
err = None if 200 <= code < 500 else (r2.stderr.decode()[:80] or "curl_fail")
|
||
out = {
|
||
"ok": 200 <= code < 500,
|
||
"http": code,
|
||
"ms": ms,
|
||
"bytes": len(body.encode("utf-8", "replace")),
|
||
"error": None if 200 <= code < 500 else err,
|
||
"tool": "curl",
|
||
"snippet": body[:80].replace("\n", " ") if body else "",
|
||
}
|
||
if keep_body and body:
|
||
out["body"] = body[:body_limit]
|
||
return out
|
||
return {
|
||
"ok": False, "http": 0, "ms": ms, "bytes": 0,
|
||
"error": (r.stderr.decode("utf-8", "replace") or "curl_fail")[:120],
|
||
"tool": "curl",
|
||
}
|
||
out = {
|
||
"ok": 200 <= code < 500,
|
||
"http": code,
|
||
"ms": ms,
|
||
"bytes": len(body.encode("utf-8", "replace")),
|
||
"error": None if 200 <= code < 500 else f"http_{code}",
|
||
"tool": "curl",
|
||
"snippet": body[:80].replace("\n", " ") if body else "",
|
||
}
|
||
if keep_body and body:
|
||
out["body"] = body[:body_limit]
|
||
return out
|
||
except Exception as e:
|
||
return {"ok": False, "http": 0, "ms": int((time.monotonic() - t0) * 1000),
|
||
"bytes": 0, "error": str(e)[:120], "tool": "curl"}
|
||
|
||
# urllib fallback（清网直连）
|
||
t0 = time.monotonic()
|
||
req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "*/*"})
|
||
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
try:
|
||
with opener.open(req, timeout=timeout) as r:
|
||
body_b = r.read(body_limit if keep_body else 6000)
|
||
body = body_b.decode("utf-8", "replace") if keep_body else ""
|
||
code = int(getattr(r, "status", 200) or 200)
|
||
ms = int((time.monotonic() - t0) * 1000)
|
||
out = {
|
||
"ok": 200 <= code < 500, "http": code, "ms": ms,
|
||
"bytes": len(body_b), "error": None, "tool": "urllib",
|
||
"snippet": (body or body_b[:80].decode("utf-8", "replace")).replace("\n", " ")[:80],
|
||
}
|
||
if keep_body and body:
|
||
out["body"] = body
|
||
return out
|
||
except urllib.error.HTTPError as e:
|
||
ms = int((time.monotonic() - t0) * 1000)
|
||
code = int(e.code)
|
||
return {
|
||
"ok": 200 <= code < 500, "http": code, "ms": ms, "bytes": 0,
|
||
"error": f"http_{code}", "tool": "urllib",
|
||
}
|
||
except Exception as e:
|
||
ms = int((time.monotonic() - t0) * 1000)
|
||
return {"ok": False, "http": 0, "ms": ms, "bytes": 0, "error": str(e)[:120], "tool": "urllib"}
|
||
|
||
|
||
# 合法/观察向 onion 基线（探活 + 标题级，不爬店内商品、不成交）
|
||
LEGAL_ONION_OBSERVE: list[dict[str, str]] = [
|
||
{
|
||
"id": "torproject-onion",
|
||
"label": "Tor Project",
|
||
"onion": "http://2gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion/",
|
||
"kind": "legal",
|
||
},
|
||
{
|
||
"id": "ddg-onion",
|
||
"label": "DuckDuckGo onion",
|
||
"onion": "https://duckduckgogg42xjoc72x3sjasowoarfbgcmvfimaftt6twagswzczad.onion/",
|
||
"kind": "legal",
|
||
},
|
||
]
|
||
|
||
|
||
def _extract_title(body: str) -> str:
|
||
if not body:
|
||
return ""
|
||
m = re.search(r"<title[^>]*>([^<]{1,200})</title>", body, flags=re.I | re.S)
|
||
if m:
|
||
return re.sub(r"\s+", " ", m.group(1)).strip()[:120]
|
||
# 纯文本首行
|
||
for ln in body.splitlines():
|
||
t = ln.strip()
|
||
if len(t) >= 4:
|
||
return t[:120]
|
||
return ""
|
||
|
||
|
||
def _load_onion_markets() -> list[dict[str, Any]]:
|
||
loaded: list[dict[str, Any]] = []
|
||
for p in (ONION_CFG, ONION_CFG_ALT, ONION_EX):
|
||
if not p.is_file():
|
||
continue
|
||
try:
|
||
d = json.loads(p.read_text(encoding="utf-8"))
|
||
mk = d.get("markets") if isinstance(d, dict) else None
|
||
if isinstance(mk, list):
|
||
loaded = [x for x in mk if isinstance(x, dict) and (x.get("onion") or x.get("url"))]
|
||
break
|
||
except (OSError, json.JSONDecodeError):
|
||
continue
|
||
# 合法基线始终在前（用户白名单可覆盖同 id）
|
||
by_id = {str(x.get("id") or ""): x for x in LEGAL_ONION_OBSERVE}
|
||
for x in loaded:
|
||
iid = str(x.get("id") or x.get("onion") or "")[:40]
|
||
if iid and "example.onion" not in str(x.get("onion") or ""):
|
||
by_id[iid] = x
|
||
return list(by_id.values())
|
||
|
||
|
||
def _load_shops() -> dict[str, Any]:
|
||
for p in SHOPS_PATHS:
|
||
if not p.is_file():
|
||
continue
|
||
try:
|
||
d = json.loads(p.read_text(encoding="utf-8"))
|
||
if isinstance(d, dict):
|
||
shops = [x for x in (d.get("shops") or []) if isinstance(x, dict) and x.get("id")]
|
||
return {
|
||
"path": str(p),
|
||
"shops": shops,
|
||
"brands": list(d.get("brands") or []),
|
||
"rsshub_base": (d.get("rsshub_base") or "").strip(),
|
||
"n": len(shops),
|
||
}
|
||
except (OSError, json.JSONDecodeError):
|
||
continue
|
||
return {"path": None, "shops": [], "brands": [], "rsshub_base": "", "n": 0}
|
||
|
||
|
||
def _brands(shops_meta: dict[str, Any]) -> list[str]:
|
||
env = (os.environ.get("BRAND_DOMAINS") or "").strip()
|
||
out: list[str] = []
|
||
if env:
|
||
out.extend([x.strip().lower() for x in re.split(r"[,;\s]+", env) if x.strip()])
|
||
for b in shops_meta.get("brands") or []:
|
||
if isinstance(b, str) and b.strip():
|
||
out.append(b.strip().lower())
|
||
out.extend(DEFAULT_BRANDS)
|
||
# de-dupe keep order
|
||
seen: set[str] = set()
|
||
uniq: list[str] = []
|
||
for b in out:
|
||
if b not in seen:
|
||
seen.add(b)
|
||
uniq.append(b)
|
||
return uniq
|
||
|
||
|
||
def _brand_severity(source: str) -> tuple[str, str]:
|
||
"""返回 (severity, kind)。threat=钓鱼/恶意 feed；exposure=自有域 urlscan 等旁证。"""
|
||
s = (source or "").lower()
|
||
if any(x in s for x in ("openphish", "phishing", "maltrail", "urlhaus", "threatfox")):
|
||
return "threat", "threat"
|
||
if s.startswith("urlscan-brand") or "urlscan" in s:
|
||
return "exposure", "exposure"
|
||
if "ransom" in s or "hibp" in s:
|
||
return "watch", "intel"
|
||
return "info", "info"
|
||
|
||
|
||
def _scan_brands(body: str, brands: list[str], *, source: str, max_hits: int = 12) -> list[dict[str, Any]]:
|
||
if not body or not brands:
|
||
return []
|
||
low = body.lower()
|
||
sev, kind = _brand_severity(source)
|
||
hits: list[dict[str, Any]] = []
|
||
for b in brands:
|
||
if not b or b not in low:
|
||
continue
|
||
# collect up to 2 context lines / urls containing brand
|
||
contexts: list[str] = []
|
||
for line in body.splitlines():
|
||
if b in line.lower():
|
||
contexts.append(line.strip()[:160])
|
||
if len(contexts) >= 2:
|
||
break
|
||
hits.append({
|
||
"brand": b,
|
||
"source": source,
|
||
"severity": sev,
|
||
"kind": kind,
|
||
"n_lines": sum(1 for ln in body.splitlines() if b in ln.lower()),
|
||
"samples": contexts,
|
||
})
|
||
if len(hits) >= max_hits:
|
||
break
|
||
return hits
|
||
|
||
|
||
def _sample_ransomware(body: str) -> dict[str, Any]:
|
||
try:
|
||
data = json.loads(body)
|
||
except json.JSONDecodeError:
|
||
return {"ok": False, "n": 0}
|
||
rows = data if isinstance(data, list) else (data.get("victims") or data.get("posts") or [])
|
||
if not isinstance(rows, list):
|
||
return {"ok": True, "n": 0, "recent": []}
|
||
recent = []
|
||
for r in rows[:8]:
|
||
if not isinstance(r, dict):
|
||
continue
|
||
recent.append({
|
||
"victim": (r.get("victim") or r.get("post_title") or r.get("name") or "")[:80],
|
||
"group": (r.get("group") or r.get("group_name") or r.get("gang") or "")[:40],
|
||
"country": r.get("country") or r.get("country_code") or "",
|
||
"date": str(r.get("attackdate") or r.get("discovered") or r.get("date") or "")[:32],
|
||
})
|
||
return {"ok": True, "n": len(rows), "recent": recent}
|
||
|
||
|
||
def _sample_lines(body: str, limit: int = 5) -> list[str]:
|
||
lines = [ln.strip() for ln in body.splitlines() if ln.strip() and not ln.startswith("#")]
|
||
return lines[:limit]
|
||
|
||
|
||
def probe() -> dict[str, Any]:
|
||
_load_keys()
|
||
socks = _detect_socks()
|
||
shops_meta = _load_shops()
|
||
brands = _brands(shops_meta)
|
||
checks: list[dict[str, Any]] = []
|
||
findings: list[dict[str, Any]] = []
|
||
brand_hits: list[dict[str, Any]] = []
|
||
samples: dict[str, Any] = {}
|
||
|
||
# 1) clearnet health
|
||
for c in CLEARNET_CHECKS:
|
||
rec = _fetch(c["url"], timeout=10, direct=True)
|
||
row = {"id": c["id"], "kind": c["kind"], "url": c["url"], "via": "clearnet", **rec}
|
||
checks.append(row)
|
||
if not rec.get("ok"):
|
||
findings.append({
|
||
"level": "warn",
|
||
"id": c["id"],
|
||
"msg": f"clearnet down/fail: {c['id']} http={rec.get('http')} {rec.get('error')}",
|
||
})
|
||
time.sleep(0.12)
|
||
|
||
# 2) JSON feeds — 部分 keep_body 做品牌撞库 / 抽样
|
||
SAMPLE_IDS = {
|
||
"openphish-feed", "phishing-db-today", "urlscan-brand", "urlscan-brand-wenpai",
|
||
"ransomware-api-recent", "ransomlook-posts", "maltrail-onion", "hibp-latest",
|
||
"ransomwatch-posts",
|
||
}
|
||
for c in JSON_CHECKS:
|
||
keep = c["id"] in SAMPLE_IDS
|
||
rec = _fetch(c["url"], timeout=15, direct=True, keep_body=keep)
|
||
body = rec.pop("body", None) if isinstance(rec, dict) else None
|
||
row = {"id": c["id"], "kind": c["kind"], "url": c["url"], "via": "clearnet", **rec}
|
||
checks.append(row)
|
||
if keep and rec.get("ok") and body:
|
||
hits = _scan_brands(body, brands, source=c["id"])
|
||
if hits:
|
||
brand_hits.extend(hits)
|
||
threat_n = sum(1 for h in hits if h.get("severity") == "threat")
|
||
findings.append({
|
||
"level": "crit" if threat_n else "info",
|
||
"id": c["id"],
|
||
"msg": (
|
||
f"brand {'THREAT' if threat_n else 'exposure'} in {c['id']}: "
|
||
+ ", ".join(h["brand"] for h in hits[:6])
|
||
),
|
||
})
|
||
if c["id"] in ("ransomware-api-recent", "ransomlook-posts", "ransomwatch-posts"):
|
||
samples[c["id"]] = _sample_ransomware(body)
|
||
elif c["id"] in ("openphish-feed", "phishing-db-today", "maltrail-onion"):
|
||
samples[c["id"]] = {
|
||
"ok": True,
|
||
"lines": len(body.splitlines()),
|
||
"head": _sample_lines(body, 4),
|
||
}
|
||
elif c["id"].startswith("urlscan-brand"):
|
||
try:
|
||
j = json.loads(body)
|
||
samples[c["id"]] = {
|
||
"ok": True,
|
||
"total": (j.get("total") if isinstance(j, dict) else None),
|
||
"n": len((j.get("results") or []) if isinstance(j, dict) else []),
|
||
}
|
||
except json.JSONDecodeError:
|
||
samples[c["id"]] = {"ok": True, "bytes": len(body)}
|
||
elif c["id"] == "hibp-latest":
|
||
try:
|
||
j = json.loads(body)
|
||
samples[c["id"]] = {
|
||
"ok": True,
|
||
"Name": j.get("Name") if isinstance(j, dict) else None,
|
||
"BreachDate": j.get("BreachDate") if isinstance(j, dict) else None,
|
||
"PwnCount": j.get("PwnCount") if isinstance(j, dict) else None,
|
||
}
|
||
except json.JSONDecodeError:
|
||
pass
|
||
time.sleep(0.08)
|
||
|
||
# 2b) shops with public URL — 首页健康（跳过 example/invalid）
|
||
shop_checks = 0
|
||
for s in shops_meta.get("shops") or []:
|
||
url = (s.get("url") or "").strip()
|
||
sid = str(s.get("id") or "")
|
||
if not url or "example" in url or ".invalid" in url or "example" in sid:
|
||
continue
|
||
if not url.startswith("http"):
|
||
continue
|
||
rec = _fetch(url, timeout=10, direct=True)
|
||
checks.append({
|
||
"id": f"shop:{sid}",
|
||
"kind": "shop_home",
|
||
"url": url,
|
||
"label": s.get("name") or sid,
|
||
"via": "clearnet",
|
||
**rec,
|
||
})
|
||
shop_checks += 1
|
||
if not rec.get("ok"):
|
||
findings.append({
|
||
"level": "info",
|
||
"id": sid,
|
||
"msg": f"shop down: {s.get('name') or sid} http={rec.get('http')}",
|
||
})
|
||
time.sleep(0.1)
|
||
if shop_checks >= 12:
|
||
break
|
||
|
||
# 2c) optional RSSHub public channel titles（仅当配置了 base + path）
|
||
rsshub = (shops_meta.get("rsshub_base") or os.environ.get("RSSHUB_BASE") or "").strip().rstrip("/")
|
||
if rsshub:
|
||
for s in (shops_meta.get("shops") or [])[:8]:
|
||
path = (s.get("rsshub_path") or "").strip()
|
||
if not path:
|
||
continue
|
||
if not path.startswith("/"):
|
||
path = "/" + path
|
||
url = rsshub + path
|
||
rec = _fetch(url, timeout=12, direct=True, keep_body=True, body_limit=30_000)
|
||
body = rec.pop("body", None)
|
||
checks.append({
|
||
"id": f"rsshub:{s.get('id')}",
|
||
"kind": "rsshub",
|
||
"url": url,
|
||
"via": "clearnet",
|
||
**{k: v for k, v in rec.items() if k != "body"},
|
||
})
|
||
if rec.get("ok") and body:
|
||
titles = re.findall(r"<title>([^<]{3,120})</title>", body, flags=re.I)
|
||
samples[f"rsshub:{s.get('id')}"] = {
|
||
"ok": True,
|
||
"titles": [t.strip()[:100] for t in titles[1:6]], # skip channel title
|
||
}
|
||
time.sleep(0.15)
|
||
|
||
# 3) via Tor SOCKS if up — check.torproject + onion list
|
||
tor_ok = bool(socks.get("ok") and socks.get("proxy"))
|
||
if tor_ok:
|
||
# confirm exit (API preferred)
|
||
rec = _fetch(
|
||
"https://check.torproject.org/api/ip",
|
||
timeout=25,
|
||
proxy=socks["proxy"],
|
||
direct=False,
|
||
)
|
||
checks.append({
|
||
"id": "tor-exit-api",
|
||
"kind": "tor_exit",
|
||
"url": "https://check.torproject.org/api/ip",
|
||
"via": "socks",
|
||
**rec,
|
||
})
|
||
if not rec.get("ok"):
|
||
findings.append({
|
||
"level": "warn",
|
||
"id": "tor-exit-api",
|
||
"msg": "SOCKS open but check.torproject api/ip failed",
|
||
})
|
||
# onion 标题级观察（合法基线 + 白名单；不成交、不爬店内 SKU）
|
||
onion_titles: list[dict[str, Any]] = []
|
||
for m in _load_onion_markets():
|
||
onion = (m.get("onion") or m.get("url") or "").strip()
|
||
if not onion or "example.onion" in onion:
|
||
continue
|
||
mid = str(m.get("id") or onion[:24])
|
||
is_legal = str(m.get("kind") or "") == "legal" or mid in (
|
||
"torproject-onion", "ddg-onion", "onion:torproject",
|
||
)
|
||
rec = _fetch(
|
||
onion,
|
||
timeout=40 if is_legal else 28,
|
||
proxy=socks["proxy"],
|
||
direct=False,
|
||
keep_body=True,
|
||
body_limit=80_000,
|
||
)
|
||
body = rec.pop("body", None) if isinstance(rec, dict) else None
|
||
title = _extract_title(body or "") if rec.get("ok") and body else ""
|
||
kind = "onion_legal" if is_legal else "onion_observe"
|
||
checks.append({
|
||
"id": f"onion:{mid}",
|
||
"kind": kind,
|
||
"url": onion[:80],
|
||
"label": m.get("label"),
|
||
"title": title or None,
|
||
"via": "socks",
|
||
**{k: v for k, v in rec.items() if k != "body"},
|
||
})
|
||
onion_titles.append({
|
||
"id": mid,
|
||
"label": m.get("label") or mid,
|
||
"ok": bool(rec.get("ok")),
|
||
"title": title,
|
||
"ms": rec.get("ms"),
|
||
"legal": is_legal,
|
||
"observe_only": True,
|
||
})
|
||
if not rec.get("ok"):
|
||
findings.append({
|
||
"level": "info",
|
||
"id": mid,
|
||
"msg": f"onion unreachable: {m.get('label') or onion[:40]}",
|
||
})
|
||
elif title:
|
||
findings.append({
|
||
"level": "info",
|
||
"id": mid,
|
||
"msg": f"onion title: {(m.get('label') or mid)} · {title[:60]}",
|
||
})
|
||
time.sleep(0.25)
|
||
if onion_titles:
|
||
samples["onion_titles"] = {
|
||
"ok": True,
|
||
"n": len(onion_titles),
|
||
"alive": sum(1 for x in onion_titles if x.get("ok")),
|
||
"items": onion_titles[:16],
|
||
"policy": "observe_only_no_checkout",
|
||
}
|
||
else:
|
||
findings.append({
|
||
"level": "warn",
|
||
"id": "tor-socks",
|
||
"msg": "Tor SOCKS not listening on 9050/9150 — onion probes skipped",
|
||
})
|
||
|
||
ok_n = sum(1 for c in checks if c.get("ok"))
|
||
host = socket.gethostname()
|
||
circuit_ok = any(
|
||
c.get("id") in ("tor-exit-api", "tor-exit-check") and c.get("ok")
|
||
for c in checks
|
||
)
|
||
socks = dict(socks)
|
||
socks["circuit_ok"] = circuit_ok
|
||
if socks.get("ok") and not circuit_ok:
|
||
findings.append({
|
||
"level": "warn",
|
||
"id": "tor-circuit",
|
||
"msg": "SOCKS :9050 在听但电路不通（curl socks 超时）— 查 snowflake/网桥/出站",
|
||
})
|
||
threat_hits = [h for h in brand_hits if h.get("severity") == "threat"]
|
||
if threat_hits:
|
||
findings.insert(0, {
|
||
"level": "crit",
|
||
"id": "brand-hits-threat",
|
||
"msg": f"品牌 THREAT {len(threat_hits)} 条（钓鱼/恶意 feed）· 立即排查",
|
||
})
|
||
elif brand_hits:
|
||
findings.insert(0, {
|
||
"level": "info",
|
||
"id": "brand-hits-exposure",
|
||
"msg": f"品牌曝光旁证 {len(brand_hits)} 条（urlscan 等，非钓鱼）",
|
||
})
|
||
out = {
|
||
"ts": _now(),
|
||
"ok": True,
|
||
"host": host,
|
||
"role": "scout" if "data/manhuang" in str(OUT_DIR) else "local",
|
||
"socks": socks,
|
||
"brands": brands,
|
||
"shops": {
|
||
"path": shops_meta.get("path"),
|
||
"n": shops_meta.get("n") or 0,
|
||
"items": [
|
||
{
|
||
"id": s.get("id"),
|
||
"name": s.get("name"),
|
||
"kind": s.get("kind"),
|
||
"platform": s.get("platform"),
|
||
"handle": s.get("handle"),
|
||
"status": s.get("status"),
|
||
"sku_for": (s.get("sku_for") or [])[:6],
|
||
"has_url": bool(s.get("url")),
|
||
}
|
||
for s in (shops_meta.get("shops") or [])[:40]
|
||
if "example" not in str(s.get("id") or "")
|
||
],
|
||
},
|
||
"brand_hits": brand_hits[:30],
|
||
"samples": samples,
|
||
"counts": {
|
||
"checks": len(checks),
|
||
"ok": ok_n,
|
||
"fail": len(checks) - ok_n,
|
||
"findings": len(findings),
|
||
"onion_probed": sum(
|
||
1 for c in checks
|
||
if str(c.get("kind") or "").startswith("onion")
|
||
),
|
||
"clearnet_ok": sum(1 for c in checks if c.get("via") == "clearnet" and c.get("ok")),
|
||
"circuit_ok": circuit_ok,
|
||
"brand_hits": len(brand_hits),
|
||
"brand_threat": len(threat_hits),
|
||
"brand_exposure": sum(1 for h in brand_hits if h.get("severity") == "exposure"),
|
||
"shops": shops_meta.get("n") or 0,
|
||
"shop_probed": shop_checks,
|
||
},
|
||
"checks": checks,
|
||
"findings": findings[:40],
|
||
"paths": {
|
||
"out": str(OUT),
|
||
"onion_cfg": str(ONION_CFG),
|
||
"shops": shops_meta.get("path"),
|
||
},
|
||
}
|
||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
OUT.write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return out
|
||
|
||
|
||
def main() -> int:
|
||
global OUT_DIR, OUT
|
||
OUT_DIR = _data_dir()
|
||
OUT = OUT_DIR / "status.json"
|
||
r = probe()
|
||
c = r.get("counts") or {}
|
||
print(
|
||
"wrote", OUT,
|
||
"host", r.get("host"),
|
||
"role", r.get("role"),
|
||
"checks", c.get("checks"),
|
||
"ok", c.get("ok"),
|
||
"socks", (r.get("socks") or {}).get("ok"),
|
||
"onion", c.get("onion_probed"),
|
||
"brand_hits", c.get("brand_hits"),
|
||
"shops", c.get("shops"),
|
||
)
|
||
for f in (r.get("findings") or [])[:8]:
|
||
print("-", f.get("level"), f.get("msg"))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|
