Shafeeq-Cybersec/threat-intel-platform

# AI 驱动的安全运营与威胁情报平台 一个生产级、自托管的安全运营中心 (SOC) 平台，可并行编排多个威胁情报 pipeline，使用多源风险聚合引擎关联信号，并通过实时的 SocketIO dashboard 呈现可操作的判定结果。在商业威胁情报源之上集成了 Google Gemini 作为 AI 推理层。 ## 架构概述 ``` flowchart TD User([User / Browser]) --> Dashboard[Flask Dashboard\nSocketIO Server] Ext([Chrome Extension]) --> Dashboard Dashboard --> URL[URL Scanner] Dashboard --> QR[QR Analyzer] Dashboard --> SOC[SOC Log Analyzer] Dashboard --> Email[Email Phishing Scanner] Dashboard --> File[File Hash Scanner] Dashboard --> IP[IP / Domain Lookup] URL --> AGG[Risk Aggregation Engine\nParallel ThreadPoolExecutor] QR --> AGG AGG --> VT[VirusTotal\n70+ engines] AGG --> SB[Google Safe Browsing] AGG --> RDAP[RDAP / Whois\nDomain Age] AGG --> PT[PhishTank] SOC --> RULE[Rule Engine\nBrute Force / Off-Hours\nImpossible Travel] SOC --> ABIP[AbuseIPDB\nIP Reputation] Email --> NLP[NLP Heuristics\nUrgency / Sender / URLs] Email --> URLS[Threaded URL Scan] VT --> GEMINI[Gemini AI Reasoning\n3-Key Rotation Pool\n429 Auto-Failover] SB --> GEMINI RULE --> GEMINI NLP --> GEMINI ABIP --> GEMINI GEMINI --> MONGO[(MongoDB Atlas\nPersistent Storage\n24h Deduplication)] URLS --> MONGO MONGO --> FEED[Live SocketIO Feed] MONGO --> ALERT[Brevo Email Alert] MONGO --> MAP[Threat Map\nGeoIP Clustering] MONGO --> REPORT[PDF Report] FEED --> Dashboard ``` ## 功能矩阵 | 模块 | 检测 Pipeline | 存储 | 实时 | |---|---|---|---| | URL 扫描器 | VT (70+ 引擎) + SafeBrowsing + RDAP age + Gemini 推理 | MongoDB `threat_logs` | SocketIO 广播 | | 二维码分析器 | OpenCV 解码 → 完整 URL pipeline | MongoDB `threat_logs` | SocketIO 广播 | | SOC 日志分析器 | Regex 规则引擎 + AbuseIPDB IP 信誉 + Gemini 歧义处理 | MongoDB `soc_events` | SocketIO 广播 | | 邮件钓鱼扫描器 | 发件人域信任评分 + URL 信誉 (多线程) + 紧急性 NLP 启发式 + Gemini 判定 | MongoDB `email_scans` | SocketIO 广播 | | IP/域名情报 | RDAP, 地理定位, ASN, 滥用置信度 | - | - | | 文件哈希扫描器 | SHA-256/MD5/SHA-1 多哈希 + VirusTotal 查询 | - | - | | 威胁地图 | 跨所有扫描类型的攻击者 IP 的 GeoIP 聚合 | - | 实时聚类 | | Chrome 扩展 | 命中实时后端 API 的后台页面扫描器 | - | - | ## 技术栈 | 层级 | 技术 | 原因 | |---|---|---| | Runtime | Python 3.11 + eventlet | eventlet 对标准库进行 monkey-patch 以实现异步 I/O，无需全面重写为 ASGI | | Web 框架 | Flask + Flask-SocketIO | 基于 Blueprint 的按功能划分架构，保持每个模块可独立测试 | | 异步 worker | Gunicorn + eventlet worker 类 | 单线程事件循环；避免与 pymongo 发生线程冲突 | | AI | Google Gemini (`gemini-2.0-flash-lite`, `gemini-2.5-flash`) | 根据上下文大小和延迟要求，在不同模块中使用不同的模型 | | 密钥管理 | 带有 429 自动故障转移的 3 密钥轮换池 | 在无任何付费计划的情况下，将免费层级的每日配额延长 3 倍 | | 数据库 | MongoDB Atlas (云端) | 抵御短暂的 Render 文件系统重置；按时间戳 + source_type 建立索引 | | 去重 | 每个 (source_type, input_value) 24 小时窗口 | 防止因重复的相同扫描导致 dashboard 数据虚高 | | 二维码解码 | OpenCV `cv2.QRCodeDetector` | 与 pyzbar (需要 libiconv/zbar) 相比，无需任何原生系统依赖 | | 邮件告警 | Brevo REST API | Render 免费层级封锁了 SMTP (端口 465/587 被防火墙拦截) | | 实时推送 | Socket.IO (基于 eventlet 异步) | 每次检测到威胁时，向所有已连接客户端推送的延迟低于 100ms | | 前端 | 原生 HTML/CSS/JS + Socket.IO 客户端 | 无需构建步骤；模板直接由 Flask 提供 | | 活跃检测 | UptimeRobot 5 分钟 ping | 防止 Render 免费层级进入 15 分钟的休眠状态 | ## 项目结构 ``` . ├── backend/ │ ├── app.py # Flask app factory, blueprint registration, SocketIO init │ ├── config.py # Env loader, Gemini key rotation state, is_configured() guard │ ├── realtime.py # SocketIO broadcast helper + severity score map │ ├── requirements.txt │ ├── runtime.txt # Pins Python 3.11.9 for Render (prevents 3.14 auto-select) │ ├── models/ │ │ └── database.py # MongoDB client (lazy init), insert/read helpers, 24h dedup │ ├── routes/ # One Flask Blueprint per feature │ │ ├── check_url.py # /api/check-url, bulk scan, 24h cache layer │ │ ├── scan_qr.py # /api/scan-qr, OpenCV decode + URL pipeline │ │ ├── soc_analyzer.py # /api/analyze-logs, rule engine + AI pass │ │ ├── email_scanner.py # /email_scanner/api/scan + scan-eml, dedup by sender+subject │ │ ├── geo.py # /api/geo-threats, GeoIP aggregation across collections │ │ ├── settings.py # /api/stats, /api/feed, /api/history, /api/export CSV │ │ ├── ip_lookup.py # /api/ip-lookup │ │ ├── file_scan.py # /api/file-scan │ │ └── dashboard.py # / → dashboard.html │ ├── services/ # One file per external integration (all gracefully degrade) │ │ ├── risk_aggregator.py # Parallel ThreadPoolExecutor fan-out + score fusion │ │ ├── gemini_service.py # Gemini reasoning with 3-key rotation + fallback │ │ ├── virustotal_service.py │ │ ├── safebrowsing_service.py │ │ ├── abuseipdb_service.py │ │ ├── rdap_service.py │ │ ├── phishtank_service.py │ │ ├── email_phishing_service.py # NLP heuristics + threaded URL scan + Gemini verdict │ │ ├── log_analyzer.py # Regex rule engine + AI-assisted ambiguous line detection │ │ ├── alert_service.py # Brevo REST alert with risk-band HTML email │ │ ├── qr_service.py │ │ ├── geo_service.py │ │ ├── lookup_service.py │ │ ├── report_service.py # FPDF2 PDF report generation │ │ └── mail_service.py │ ├── templates/ │ │ └── dashboard.html # Single-page app, SocketIO client, all feature UIs │ └── sample_data/ # Sample QR images and log files for demo ├── chrome-extension/ │ ├── manifest.json # MV3 extension manifest │ ├── popup.html / popup.js # Extension UI + backend API calls │ └── background.js ├── .env.example # All configurable environment variables documented └── render.yaml # Render deployment config ``` ## 环境变量 ``` # Gemini AI (3-key pool 用于 quota rotation) GEMINI_API_KEY= GEMINI_API_KEY_2= GEMINI_API_KEY_3= # Threat Intelligence APIs VIRUSTOTAL_API_KEY= GOOGLE_SAFE_BROWSING_API_KEY= ABUSEIPDB_API_KEY= PHISHTANK_API_KEY= # MongoDB Atlas MONGO_URI=mongodb+srv://... # Email Alerting (Brevo REST) BREVO_API_KEY= ALERT_EMAIL_FROM= ALERT_EMAIL_TO= ALERT_THRESHOLD=70 # App SECRET_KEY= DEBUG=false ``` 所有集成都能优雅降级。平台在无任何密钥的情况下也能运行，仅使用 RDAP 和启发式算法。每个密钥都会解锁一个额外的检测层。 ## 本地设置 ``` # 要求精确使用 Python 3.11 (eventlet 0.37.0 与 3.12+ 不兼容) python3.11 -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate pip install -r backend/requirements.txt cp .env.example .env # Fill in keys cd backend && python app.py ``` 打开 `http://localhost:5000` ## 部署 部署在 Render 上 (Python 3.11.9, gunicorn + eventlet worker)。数据库使用 MongoDB Atlas 免费层级 (M0)。UptimeRobot 每 5 分钟进行一次 ping 操作，以防止冷启动。 ``` gunicorn -k eventlet -w 1 --timeout 120 --bind 0.0.0.0:$PORT app:app ``` 解决的关键部署限制： - Render 免费层级封锁了 SMTP (465/587) —— 通过 Brevo REST API 解决 - Render 文件系统是临时的 —— 通过使用 MongoDB Atlas 存储所有扫描记录来解决 - Gemini 免费层级配额耗尽 —— 通过带有 429 自动故障转移的 3 密钥轮询调度来解决 - Python 版本不匹配 (Render 默认为 3.14) —— 通过在根目录下放置 `runtime.txt` 来解决 ## 检测逻辑 ### URL 风险评分融合 每个数据源都会返回一个 0-100 的分数。聚合器在并行线程中运行所有数据源并融合结果： ``` final_score = weighted_average([vt, safebrowsing, phishtank, rdap_age_penalty]) → passed to Gemini with evidence JSON for contextual reasoning → Gemini returns: risk_score, threat_label, reasoning, contributing_factors[] ``` ### SOC 日志分析 (两步处理) **第一步 —— 规则引擎：** - 暴力破解：来自同一 IP 的 5 次以上失败登录 - 非工作时间访问：00:00-05:59 之间的成功登录 - 已知的恶意 IP：AbuseIPDB 置信度分数 - 不可能完成的旅行：同一用户在 2 个以上的国家/地区登录 **第二步 —— AI 处理 (上限为 8 行)：** - 匹配关键词 (sudo, privilege, escalat, exploit...) 的歧义行发送给 Gemini - 返回：每个可疑行的 line_index, severity, category, explanation ### 邮件钓鱼 Pipeline ``` 1. Sender domain trust check (whitelist of 30+ known providers) 2. URL extraction → parallel VirusTotal/SafeBrowsing scan (max 5 URLs, threaded) 3. NLP urgency heuristics (12 regex patterns: account suspension, verify identity, etc.) 4. Reply-To domain mismatch detection 5. Risky attachment extension check (.exe, .js, .vbs, .iso, .lnk, .docm...) 6. Raw IP link detection 7. Gemini final verdict with full evidence bundle ``` ## Chrome 扩展 MV3 扩展。读取当前标签页的 URL 并将其发送到配置的后端。在弹窗中内联显示风险评分、威胁标签和数据源。按 URL 缓存结果。 在扩展设置中配置后端 URL (默认：`http://localhost:5000`)。

标签：IP 地址批量处理, SOC平台, 后端开发, 威胁情报, 安全运营中心, 开发者工具, 异常检测, 网络安全, 网络映射, 隐私保护, 风险聚合引擎