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.
421 lines
14 KiB
Python
421 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""Correlate open feeds against own scope/stack → hits + actions.
|
||
|
||
Appends to open-summary.json:
|
||
hits: { brand_ioc, kev_stack, cc_history, phishing_lookalike }
|
||
actions: [{priority, kind, title, detail, next, evidence[]}]
|
||
scope / stack meta
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse
|
||
|
||
OUT = Path.home() / "data/manhuang/feeds/open"
|
||
SUMMARY = Path.home() / "data/manhuang/feeds/open-summary.json"
|
||
RECENT_KEV = Path.home() / "data/manhuang/feeds/cisa-kev-recent.json"
|
||
|
||
# Own surface
|
||
DOMAINS = (
|
||
"feibisi.com",
|
||
"weixiaoduo.com",
|
||
"cravatar.com",
|
||
"wenpai.org",
|
||
"feicode.com",
|
||
)
|
||
BRAND_TOKENS = (
|
||
"feibisi", "weixiaoduo", "cravatar", "wenpai", "feicode",
|
||
"菲比斯", "文派", "微晓多",
|
||
)
|
||
|
||
# Stack we actually care about (WP 生态 + 自建)
|
||
STACK = (
|
||
("wordpress", "WordPress"),
|
||
("woocommerce", "WooCommerce"),
|
||
("elementor", "Elementor"),
|
||
("php", "PHP"),
|
||
("nginx", "nginx"),
|
||
("apache", "Apache"),
|
||
("openssl", "OpenSSL"),
|
||
("openssh", "OpenSSH"),
|
||
("linux", "Linux kernel"),
|
||
("docker", "Docker"),
|
||
("joomla", "Joomla"),
|
||
("langflow", "Langflow"),
|
||
("mysql", "MySQL"),
|
||
("mariadb", "MariaDB"),
|
||
("redis", "Redis"),
|
||
("nodejs", "Node.js"),
|
||
("python", "Python"),
|
||
("curl", "curl"),
|
||
("git", "Git"),
|
||
("kubernetes", "Kubernetes"),
|
||
("chrome", "Chrome/Chromium"),
|
||
)
|
||
|
||
|
||
def _now() -> str:
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
def _host(url: str) -> str:
|
||
try:
|
||
h = (urlparse(url).hostname or "").lower()
|
||
return h[4:] if h.startswith("www.") else h
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _match_domain(text: str) -> list[str]:
|
||
t = (text or "").lower()
|
||
hit = []
|
||
for d in DOMAINS:
|
||
if d in t:
|
||
hit.append(d)
|
||
return hit
|
||
|
||
|
||
def _match_brand_token(text: str) -> list[str]:
|
||
t = text or ""
|
||
tl = t.lower()
|
||
hit = []
|
||
for tok in BRAND_TOKENS:
|
||
if tok.isascii():
|
||
if tok in tl:
|
||
hit.append(tok)
|
||
else:
|
||
if tok in t:
|
||
hit.append(tok)
|
||
return hit
|
||
|
||
|
||
def scan_url_list(path: Path, source: str, limit_scan: int = 200000) -> list[dict]:
|
||
"""Return brand/domain hits from a URL-per-line or CSV file."""
|
||
if not path.is_file():
|
||
return []
|
||
hits = []
|
||
seen = set()
|
||
n = 0
|
||
for raw in path.read_text(errors="ignore").splitlines():
|
||
n += 1
|
||
if n > limit_scan:
|
||
break
|
||
line = raw.strip()
|
||
if not line or line.startswith("#") or line.startswith(";"):
|
||
continue
|
||
# urlhaus csv: "id","date","url",...
|
||
if "," in line and line[0] == '"':
|
||
parts = line.split(",")
|
||
url = parts[2].strip().strip('"') if len(parts) >= 3 else line
|
||
else:
|
||
url = line.split()[0] if line.split() else line
|
||
if not url.startswith("http") and "." not in url:
|
||
continue
|
||
blob = url
|
||
doms = _match_domain(blob)
|
||
brands = _match_brand_token(blob) if not doms else []
|
||
if not doms and not brands:
|
||
# lookalike: brand token in host but not exact domain
|
||
host = _host(url if url.startswith("http") else "http://" + url)
|
||
brands = _match_brand_token(host)
|
||
if not brands:
|
||
continue
|
||
kind = "lookalike"
|
||
else:
|
||
kind = "brand_ioc" if doms else "lookalike"
|
||
key = (source, url[:120])
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
hits.append({
|
||
"kind": kind,
|
||
"source": source,
|
||
"url": url[:200],
|
||
"host": _host(url if url.startswith("http") else "http://" + url),
|
||
"domains": doms,
|
||
"brands": brands,
|
||
"level": "crit" if doms else "warn",
|
||
})
|
||
if len(hits) >= 40:
|
||
break
|
||
return hits
|
||
|
||
|
||
def scan_kev_stack(path: Path) -> list[dict]:
|
||
if not path.is_file():
|
||
return []
|
||
try:
|
||
data = json.loads(path.read_text())
|
||
except Exception:
|
||
return []
|
||
vulns = data.get("vulnerabilities") or []
|
||
# prefer recent 90 days-ish by sorting dateAdded
|
||
vulns = sorted(vulns, key=lambda v: v.get("dateAdded") or "", reverse=True)
|
||
hits = []
|
||
for v in vulns[:400]: # scan recent-ish catalog slice
|
||
vendor = (v.get("vendorProject") or "")
|
||
product = (v.get("product") or "")
|
||
name = (v.get("vulnerabilityName") or "")
|
||
blob = f"{vendor} {product} {name}".lower()
|
||
matched = []
|
||
for key, label in STACK:
|
||
# word-ish match
|
||
if key == "php" and not re.search(r"\bphp\b", blob):
|
||
continue
|
||
if key == "git" and not re.search(r"\bgit\b", blob):
|
||
continue
|
||
if key == "curl" and "curl" not in blob:
|
||
continue
|
||
if key not in blob and label.lower() not in blob:
|
||
continue
|
||
matched.append(label)
|
||
if not matched:
|
||
continue
|
||
hits.append({
|
||
"kind": "kev_stack",
|
||
"source": "cisa-kev",
|
||
"cve": v.get("cveID"),
|
||
"vendor": vendor,
|
||
"product": product,
|
||
"name": name[:100],
|
||
"dateAdded": v.get("dateAdded"),
|
||
"stack": matched,
|
||
"level": "warn",
|
||
"ransomware": (v.get("knownRansomwareCampaignUse") or "").lower() == "known",
|
||
})
|
||
if len(hits) >= 25:
|
||
break
|
||
return hits
|
||
|
||
|
||
def scan_cc() -> list[dict]:
|
||
hits = []
|
||
for dom in DOMAINS:
|
||
p = OUT / f"cc-cdx-{dom}.ndjson"
|
||
if not p.is_file() or p.stat().st_size < 5:
|
||
hits.append({
|
||
"kind": "cc_history",
|
||
"source": "commoncrawl",
|
||
"domain": dom,
|
||
"n": 0,
|
||
"level": "info",
|
||
"status": "empty_or_fail",
|
||
"sample": [],
|
||
})
|
||
continue
|
||
samples = []
|
||
n = 0
|
||
for line in p.read_text(errors="ignore").splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
n += 1
|
||
if len(samples) < 3:
|
||
try:
|
||
o = json.loads(line)
|
||
samples.append({
|
||
"url": (o.get("url") or "")[:100],
|
||
"ts": (o.get("timestamp") or "")[:14],
|
||
"status": o.get("status"),
|
||
})
|
||
except Exception:
|
||
pass
|
||
hits.append({
|
||
"kind": "cc_history",
|
||
"source": "commoncrawl",
|
||
"domain": dom,
|
||
"n": n,
|
||
"level": "info",
|
||
"status": "ok" if n else "empty",
|
||
"sample": samples,
|
||
})
|
||
return hits
|
||
|
||
|
||
def build_actions(brand_hits, kev_hits, cc_hits) -> list[dict]:
|
||
actions = []
|
||
# brand IoCs first
|
||
by_dom: dict[str, list] = {}
|
||
for h in brand_hits:
|
||
if h.get("kind") != "brand_ioc":
|
||
continue
|
||
for d in h.get("domains") or []:
|
||
by_dom.setdefault(d, []).append(h)
|
||
for d, items in by_dom.items():
|
||
actions.append({
|
||
"priority": 1,
|
||
"kind": "brand_ioc",
|
||
"level": "crit",
|
||
"title": f"品牌域名出现在威胁情报：{d}",
|
||
"detail": f"{len(items)} 条来自 " + ", ".join(sorted({i['source'] for i in items})),
|
||
"next": f"立刻核验：1) 是否钓鱼仿冒 2) 是否自家被挂马 URL 3) Google Safe Browsing / urlscan 查 {d}",
|
||
"evidence": [{"source": i["source"], "url": i.get("url"), "host": i.get("host")} for i in items[:5]],
|
||
"domain": d,
|
||
})
|
||
lookalikes = [h for h in brand_hits if h.get("kind") == "lookalike"]
|
||
if lookalikes:
|
||
actions.append({
|
||
"priority": 2,
|
||
"kind": "lookalike",
|
||
"level": "warn",
|
||
"title": f"疑似品牌仿冒域名 {len(lookalikes)} 条",
|
||
"detail": ", ".join(sorted({h.get("host") or "" for h in lookalikes})[:8]),
|
||
"next": "登记仿冒域名；视情况投诉注册商 / 加入自有 block 监控",
|
||
"evidence": [{"source": h["source"], "url": h.get("url"), "host": h.get("host")} for h in lookalikes[:8]],
|
||
})
|
||
# KEV stack — group by stack label
|
||
if kev_hits:
|
||
ransomware = [k for k in kev_hits if k.get("ransomware")]
|
||
top = kev_hits[:8]
|
||
stacks = sorted({s for k in kev_hits for s in (k.get("stack") or [])})
|
||
actions.append({
|
||
"priority": 2 if not ransomware else 1,
|
||
"kind": "kev_stack",
|
||
"level": "crit" if ransomware else "warn",
|
||
"title": f"KEV 命中技术栈 {len(kev_hits)} 条（{', '.join(stacks[:6])}）",
|
||
"detail": "; ".join(
|
||
f"{k.get('cve')} {k.get('vendor')}/{k.get('product')}" for k in top[:5]
|
||
),
|
||
"next": "对照自有主机/容器版本：是否在跑对应产品；有则排补丁窗口，无则忽略并标记已评估",
|
||
"evidence": [
|
||
{
|
||
"cve": k.get("cve"),
|
||
"product": f"{k.get('vendor')}/{k.get('product')}",
|
||
"stack": k.get("stack"),
|
||
"dateAdded": k.get("dateAdded"),
|
||
"ransomware": k.get("ransomware"),
|
||
}
|
||
for k in top
|
||
],
|
||
})
|
||
# CC — only action if fail or zero when expected
|
||
fails = [c for c in cc_hits if c.get("status") != "ok"]
|
||
ok_n = sum(c.get("n") or 0 for c in cc_hits if c.get("status") == "ok")
|
||
if fails:
|
||
actions.append({
|
||
"priority": 3,
|
||
"kind": "cc_history",
|
||
"level": "info",
|
||
"title": f"Common Crawl 部分域名无快照（{len(fails)}）",
|
||
"detail": ", ".join(c["domain"] for c in fails),
|
||
"next": "多为索引失败或新域；下轮 cron 重试。无需紧急处理。",
|
||
"evidence": fails,
|
||
})
|
||
elif ok_n:
|
||
actions.append({
|
||
"priority": 4,
|
||
"kind": "cc_history",
|
||
"level": "info",
|
||
"title": f"自有域名 CC 历史基线 OK（{ok_n} 条样本）",
|
||
"detail": ", ".join(f"{c['domain']}={c['n']}" for c in cc_hits),
|
||
"next": "仅作暴露面基线；突变时再查。无需动作。",
|
||
"evidence": [{"domain": c["domain"], "n": c["n"]} for c in cc_hits],
|
||
})
|
||
# no brand hits = healthy signal
|
||
if not by_dom and not lookalikes:
|
||
actions.append({
|
||
"priority": 5,
|
||
"kind": "clear",
|
||
"level": "ok",
|
||
"title": "公开威胁源未命中自有品牌域名",
|
||
"detail": "OpenPhish / URLhaus / DigitalSide 等本轮扫描无 feibisi/wenpai/… 直接命中",
|
||
"next": "保持 6h 拉取；有命中再升级处理。",
|
||
"evidence": [],
|
||
})
|
||
actions.sort(key=lambda a: a.get("priority", 99))
|
||
return actions
|
||
|
||
|
||
def main() -> int:
|
||
brand: list[dict] = []
|
||
for fname, src in (
|
||
("openphish.txt", "openphish"),
|
||
("urlhaus-recent.csv", "urlhaus-csv"),
|
||
("digitalside-urls.txt", "digitalside-urls"),
|
||
("digitalside-domains.txt", "digitalside-domains"),
|
||
):
|
||
brand.extend(scan_url_list(OUT / fname, src))
|
||
|
||
kev_path = OUT / "cisa-kev.json"
|
||
kev = scan_kev_stack(kev_path)
|
||
cc = scan_cc()
|
||
actions = build_actions(brand, kev, cc)
|
||
|
||
hits = {
|
||
"brand_ioc": [h for h in brand if h.get("kind") == "brand_ioc"],
|
||
"lookalike": [h for h in brand if h.get("kind") == "lookalike"],
|
||
"kev_stack": kev,
|
||
"cc_history": cc,
|
||
"counts": {
|
||
"brand_ioc": sum(1 for h in brand if h.get("kind") == "brand_ioc"),
|
||
"lookalike": sum(1 for h in brand if h.get("kind") == "lookalike"),
|
||
"kev_stack": len(kev),
|
||
"cc_domains_ok": sum(1 for c in cc if c.get("status") == "ok"),
|
||
"actions": len(actions),
|
||
"actions_actionable": sum(1 for a in actions if a.get("priority", 9) <= 3 and a.get("kind") != "clear"),
|
||
},
|
||
}
|
||
|
||
# action-derived signals for live feed (only high priority)
|
||
act_signals = []
|
||
for a in actions:
|
||
if a.get("priority", 9) > 3:
|
||
continue
|
||
if a.get("kind") in ("clear",):
|
||
continue
|
||
act_signals.append({
|
||
"ts": _now()[:10],
|
||
"layer": "威胁" if a.get("kind") in ("brand_ioc", "lookalike", "kev_stack") else "深网",
|
||
"source": a.get("kind"),
|
||
"level": a.get("level") if a.get("level") in ("crit", "warn", "info") else "warn",
|
||
"title": a.get("title"),
|
||
"detail": (a.get("next") or "")[:100],
|
||
})
|
||
|
||
summary = {}
|
||
if SUMMARY.is_file():
|
||
try:
|
||
summary = json.loads(SUMMARY.read_text())
|
||
except Exception:
|
||
summary = {}
|
||
if not isinstance(summary, dict):
|
||
summary = {}
|
||
summary["ts_hits"] = _now()
|
||
summary["scope"] = {"domains": list(DOMAINS), "brands": list(BRAND_TOKENS)}
|
||
summary["stack"] = [label for _, label in STACK]
|
||
summary["hits"] = hits
|
||
summary["actions"] = actions
|
||
# prepend action signals, keep other signals but cap
|
||
old_signals = [s for s in (summary.get("signals") or []) if isinstance(s, dict)]
|
||
# drop previous auto action signals
|
||
old_signals = [s for s in old_signals if s.get("source") not in ("brand_ioc", "lookalike", "kev_stack", "cc_history")]
|
||
summary["signals"] = (act_signals + old_signals)[:40]
|
||
summary.setdefault("counts", {})
|
||
summary["counts"]["hits_brand"] = hits["counts"]["brand_ioc"]
|
||
summary["counts"]["hits_lookalike"] = hits["counts"]["lookalike"]
|
||
summary["counts"]["hits_kev_stack"] = hits["counts"]["kev_stack"]
|
||
summary["counts"]["actions"] = hits["counts"]["actions"]
|
||
summary["counts"]["actions_actionable"] = hits["counts"]["actions_actionable"]
|
||
summary["counts"]["signals"] = len(summary["signals"])
|
||
|
||
SUMMARY.write_text(json.dumps(summary, ensure_ascii=False, indent=2))
|
||
print(
|
||
"hits ok brand=%s lookalike=%s kev_stack=%s actions=%s actionable=%s"
|
||
% (
|
||
hits["counts"]["brand_ioc"],
|
||
hits["counts"]["lookalike"],
|
||
hits["counts"]["kev_stack"],
|
||
hits["counts"]["actions"],
|
||
hits["counts"]["actions_actionable"],
|
||
)
|
||
)
|
||
for a in actions[:6]:
|
||
print(" -", a.get("priority"), a.get("level"), a.get("title"))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|
