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.
183 lines
6.2 KiB
Bash
Executable file
183 lines
6.2 KiB
Bash
Executable file
#!/usr/bin/env bash
|
||
# 鬼市品牌撞库 HIT → 通知 konqi（vm-dm / office inbox）
|
||
# 只推 threat 级（钓鱼/恶意 feed 命中）；urlscan 自有域曝光默认不推（--exposure 可开）
|
||
# 节流：同 fingerprint 4h 内不重复
|
||
#
|
||
# 用法：
|
||
# bash notify-guishi-brand.sh
|
||
# bash notify-guishi-brand.sh --force
|
||
# bash notify-guishi-brand.sh --exposure # 也推 urlscan 自有域曝光
|
||
set -euo pipefail
|
||
|
||
ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
|
||
CACHE="${HOME}/.cache/manhuang"
|
||
STATUS="${CACHE}/guishi/status.json"
|
||
STAMP="${CACHE}/.guishi-brand-notify.last"
|
||
LOG="${HOME}/logs/manhuang-quotes.log"
|
||
THROTTLE_S="${GUISHI_BRAND_NOTIFY_THROTTLE_S:-14400}"
|
||
FORCE=0
|
||
EXPOSURE=0
|
||
for a in "$@"; do
|
||
case "$a" in
|
||
--force) FORCE=1 ;;
|
||
--exposure) EXPOSURE=1 ;;
|
||
esac
|
||
done
|
||
|
||
mkdir -p "$CACHE" "$(dirname "$LOG")"
|
||
log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] notify-guishi-brand: $*" | tee -a "$LOG"; }
|
||
|
||
if [[ ! -f "$STATUS" ]]; then
|
||
log "skip: no $STATUS"
|
||
exit 0
|
||
fi
|
||
|
||
export STATUS EXPOSURE
|
||
STATE=$(python3 - <<'PY'
|
||
import json, os, hashlib
|
||
from pathlib import Path
|
||
p = Path(os.environ["STATUS"])
|
||
d = json.loads(p.read_text(encoding="utf-8"))
|
||
hits = d.get("brand_hits") or []
|
||
exposure_ok = os.environ.get("EXPOSURE") == "1"
|
||
threat, exposure = [], []
|
||
for h in hits:
|
||
if not isinstance(h, dict):
|
||
continue
|
||
sev = (h.get("severity") or h.get("level") or "").lower()
|
||
src = str(h.get("source") or "")
|
||
kind = (h.get("kind") or "").lower()
|
||
is_threat = (
|
||
sev in ("threat", "crit", "high")
|
||
or kind == "threat"
|
||
or any(x in src for x in ("openphish", "phishing", "maltrail", "urlhaus", "threatfox"))
|
||
)
|
||
is_exposure = (
|
||
sev in ("exposure", "info")
|
||
or kind == "exposure"
|
||
or src.startswith("urlscan-brand")
|
||
)
|
||
if is_threat and not (is_exposure and not any(x in src for x in ("openphish", "phishing", "maltrail"))):
|
||
# urlscan alone = exposure unless marked threat
|
||
if src.startswith("urlscan-brand") and sev not in ("threat", "crit"):
|
||
exposure.append(h)
|
||
else:
|
||
threat.append(h)
|
||
elif is_exposure:
|
||
exposure.append(h)
|
||
else:
|
||
# default: treat unknown as exposure (less noise)
|
||
exposure.append(h)
|
||
|
||
notify_hits = list(threat)
|
||
if exposure_ok:
|
||
notify_hits.extend(exposure)
|
||
|
||
# fingerprint for throttle
|
||
parts = sorted({f"{h.get('brand')}|{h.get('source')}" for h in notify_hits if h.get("brand")})
|
||
fp = hashlib.sha1("|".join(parts).encode()).hexdigest()[:16] if parts else ""
|
||
|
||
# findings crit from probe
|
||
findings = [f for f in (d.get("findings") or []) if isinstance(f, dict) and f.get("level") == "crit"]
|
||
|
||
out = {
|
||
"ok": True,
|
||
"host": d.get("host") or d.get("role"),
|
||
"ts": d.get("ts"),
|
||
"threat_n": len(threat),
|
||
"exposure_n": len(exposure),
|
||
"notify_n": len(notify_hits),
|
||
"should_notify": len(notify_hits) > 0,
|
||
"fingerprint": fp,
|
||
"threat": [
|
||
{
|
||
"brand": h.get("brand"),
|
||
"source": h.get("source"),
|
||
"n_lines": h.get("n_lines"),
|
||
"sample": (h.get("samples") or [""])[0][:100],
|
||
"severity": h.get("severity") or h.get("kind"),
|
||
}
|
||
for h in notify_hits[:12]
|
||
],
|
||
"findings_crit": [f.get("msg") for f in findings[:5]],
|
||
"circuit_ok": (d.get("socks") or {}).get("circuit_ok"),
|
||
"probe_ok": (d.get("counts") or {}).get("ok"),
|
||
"probe_fail": (d.get("counts") or {}).get("fail"),
|
||
}
|
||
print(json.dumps(out, ensure_ascii=False))
|
||
PY
|
||
)
|
||
|
||
echo "$STATE" | python3 -c "import sys,json; d=json.load(sys.stdin); print('should', d.get('should_notify'), 'threat', d.get('threat_n'), 'exposure', d.get('exposure_n'), 'fp', d.get('fingerprint'))"
|
||
|
||
SHOULD=$(echo "$STATE" | python3 -c "import sys,json; print('1' if json.load(sys.stdin).get('should_notify') else '0')")
|
||
if [[ "$SHOULD" != "1" ]]; then
|
||
log "skip: no threat brand hits (use --exposure for urlscan self-domain)"
|
||
exit 0
|
||
fi
|
||
|
||
FP=$(echo "$STATE" | python3 -c "import sys,json; print(json.load(sys.stdin).get('fingerprint') or '')")
|
||
if [[ "$FORCE" != "1" && -f "$STAMP" && -n "$FP" ]]; then
|
||
now=$(date +%s)
|
||
last_epoch=$(python3 -c "import json;print(int(json.load(open('$STAMP')).get('epoch') or 0))" 2>/dev/null || echo 0)
|
||
last_fp=$(python3 -c "import json;print(json.load(open('$STAMP')).get('fingerprint') or '')" 2>/dev/null || echo "")
|
||
if [[ "$last_fp" == "$FP" && $((now - last_epoch)) -lt $THROTTLE_S ]]; then
|
||
log "skip: throttled fp=$FP age=$((now - last_epoch))s"
|
||
exit 0
|
||
fi
|
||
fi
|
||
|
||
MSG=$(echo "$STATE" | python3 -c '
|
||
import json,sys
|
||
d=json.load(sys.stdin)
|
||
lines=[
|
||
"【鬼市品牌撞库】HIT %s 条 (threat=%s exposure=%s)" % (
|
||
d.get("notify_n"), d.get("threat_n"), d.get("exposure_n")),
|
||
"host=%s probe ok/fail=%s/%s circuit=%s" % (
|
||
d.get("host"), d.get("probe_ok"), d.get("probe_fail"), d.get("circuit_ok")),
|
||
]
|
||
for h in d.get("threat") or []:
|
||
lines.append("- %s @ %s · %s" % (h.get("brand"), h.get("source"), (h.get("sample") or "")[:80]))
|
||
for f in d.get("findings_crit") or []:
|
||
lines.append("! %s" % str(f)[:100])
|
||
lines.append("面板: http://localhost:8899/#tab=mh-guishi")
|
||
print("\n".join(lines))
|
||
')
|
||
|
||
sent=0
|
||
if command -v vm-dm >/dev/null 2>&1; then
|
||
if vm-dm konqi "$MSG" --type notification --subject "manhuang-guishi-brand" 2>>"$LOG"; then
|
||
sent=1
|
||
log "sent via vm-dm konqi"
|
||
else
|
||
log "vm-dm konqi failed, try outbox"
|
||
fi
|
||
fi
|
||
|
||
if [[ "$sent" != "1" ]]; then
|
||
# konqi inbox (message-bridge) + Mac office outbox fallbacks
|
||
for OUTBOX in "${HOME}/office/konqi/inbox" "${HOME}/office/linuxjoy/outbox"; do
|
||
mkdir -p "$OUTBOX"
|
||
done
|
||
FN="${HOME}/office/konqi/inbox/manhuang-guishi-brand-$(date +%s).json"
|
||
python3 - <<PY
|
||
import json, time
|
||
from pathlib import Path
|
||
msg = '''$MSG'''
|
||
Path("$FN").write_text(json.dumps({
|
||
"from": "manhuang",
|
||
"to": "konqi",
|
||
"type": "notification",
|
||
"subject": "manhuang-guishi-brand",
|
||
"content": msg,
|
||
"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print("wrote", "$FN")
|
||
PY
|
||
log "wrote local inbox $FN"
|
||
sent=1
|
||
fi
|
||
|
||
python3 -c "import json,time; open('$STAMP','w').write(json.dumps({'epoch':time.time(),'fingerprint':'$FP','ts':time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())}))"
|
||
log "done notify_n=$(echo "$STATE" | python3 -c 'import sys,json;print(json.load(sys.stdin).get("notify_n"))')"
|
||
exit 0
|
