SukritiTkr/geosentinel

# GeoSentinel 一款用于检测登录数据中“不可能的旅行”异常的安全工具——这与 Okta、Azure AD/Entra 和 SIEM 平台等工具使用的标记潜在账户接管的技术相同。  *（注意：上面的截图是在一个没有连接互联网访问地图切片提供商的沙盒环境中截取的，因此缺少底图[海岸线/边界]。在具有互联网访问权限的普通机器上，您将在标记的路线下方看到完整的深色主题世界地图。）* ## 什么是“不可能的旅行”？ 如果一个用户从纽约登录，然后 10 分钟后又从东京登录，这在物理上是不可能的，没有人能以那么快的速度移动那么远的距离。这通常意味着账户已被攻陷（例如，凭证被盗并在其他地方使用），且有其他人正在登录该账户。 该工具会计算用户连续登录之间的隐含移动速度，并标记任何超过合理移动速度（例如，比商业航班快）的情况。 ## 威胁向量分析 ``` [Login A: New York] ─── (10 Minutes Elapsed) ───► [Login B: Tokyo] │ [Haversine Formula] │ Implied Speed: ~67,000 km/h ──► [CRITICAL ANOMALY DETECTED] ``` 1. **时间顺序排序：** 引擎会为每个单独的用户对所有入站身份验证日志进行聚类和排序。 2. **测地线计算：** 使用 **Haversine 公式**计算连续坐标之间的真实球面距离： $$d = 2R \arcsin\left(\sqrt{\sin^2\left(\frac{\Delta \phi}{2}\right) + \cos(\phi_1)\cos(\phi_2)\sin^2\left(\frac{\Delta \lambda}{2}\right)}\right)$$ 3. **速度基准：** 推导出隐含速度（$v = \text{distance} \div \text{time}$）。如果它超过了商业航空基准（$\sim 900 \text{ km/h}$），则该事件将触发警报。 4. **动态风险评分：** 根据严重性向量分配 **0-100 的风险分数**： * **速度差值：** 隐含速度违反阈值的剧烈程度。 * **时间临近度：** 较短的时间窗口（例如，5 分钟对比 4 小时）会按指数级扩大风险等级（`LOW`、`MEDIUM`、`HIGH`、`CRITICAL`）。 ## 项目状态 ✅ 已完成 — 分阶段构建： - [x] 阶段 0：项目设置 - [x] 阶段 1：合成登录数据生成器 - [x] 阶段 2：检测引擎（核心逻辑） - [x] 阶段 3：CLI 工具 - [x] 阶段 4：风险评分 - [x] 阶段 5：交互式仪表盘 - [x] 阶段 6：真实 IP 地理定位 + 优化 ## 设置 ``` python3 -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate pip install -r requirements.txt ``` ## 用法 **1. 生成示例登录数据**（创建 `data/login_events.csv`，其中植入了一些用于测试的已知“不可能的旅行”案例）： ``` python3 src/generate_data.py ``` **2. 运行检测器：** ``` python3 src/detect.py ``` 这会使用合理的默认值（读取 `data/login_events.csv`，标记任何超过 900 km/h 的事件，并将结果写入 `data/flagged_events.csv` 和 `data/detection_results.csv`）。 **使用标志自定义行为：** ``` # 使用自定义输入文件 python3 src/detect.py --input my_logs.csv # 使用更严格/更宽松的速度阈值 (km/h) python3 src/detect.py --threshold 500 # 将报告写入特定位置 python3 src/detect.py --output report.csv # 抑制每个事件的行，仅显示汇总计数 python3 src/detect.py --quiet # 查看所有可用选项 python3 src/detect.py --help ``` ## 交互式仪表盘 运行 Flask Web 应用： ``` python3 src/app.py ``` 然后在浏览器中打开 **http://127.0.0.1:5000**。您将看到： - 一个**地图**，上面以颜色编码的曲线绘制了被标记的“不可能的旅行”路线（红色 = CRITICAL，橙色 = HIGH，琥珀色 = MEDIUM，绿色 = LOW），CRITICAL 事件会呈现脉冲效果。 - 一个**侧边栏**，列出了每一个被标记的事件，按风险分数排序（最严重的排在前面）。点击任意事件即可将地图缩放到该路线。 - 顶部栏中有一个**阈值控制** — 更改速度阈值并点击 "RE-RUN DETECTION" 即可实时重新分析数据，无需重新加载页面。 仪表盘复用了与 CLI 完全相同的检测代码（`src/detect.py`）—— 无论您是从命令行还是浏览器运行它，算法的真理来源（source of truth）只有一个。 **注意：** 地图的底图切片和 Google Fonts 从外部 CDN 加载。如果您处于受限网络中（例如企业防火墙内）并且它们加载失败，仪表盘仍然可以工作——它会回退到带有参考网格的纯深色背景，并回退到系统字体。 ## 真实 IP 地理定位 阶段 1-5 使用完全合成的数据，其中已经内置了经纬度——这使得测试速度很快，但真实的登录日志通常只有一个 **IP 地址**。 `src/geolocate.py` 补充了这一缺失的部分：它使用 [ip-api.com](https://ip-api.com) 将真实的 IP 地址解析为真实的位置（免费，无需 API 密钥，免费层级约 45 次请求/分钟）。 **尝试使用一组知名的公共 IP 示例：** ``` # 1. 生成一个小型“原始”登录日志（仅包含 user、timestamp、IP -- # 无 location 列，就像真实的日志一样） python3 src/make_sample_raw_log.py # 2. 对其进行地理定位（这会对 ip-api.com 发起真实的网络调用， # 遵守 rate limit -- 预计每个 UNIQUE IP 约需 1.3s） python3 src/geolocate.py --input data/raw_logins_sample.csv --output data/raw_logins_located.csv ``` 您将得到一个 CSV，其中为每个 IP 填充了 `city`、`country`、`lat`、`lon` 列。接下来，您可以将其输入到 `detect.py` 中（在重命名/调整列以使其匹配之后），以便对真实的地理定位 IP 运行真实的检测。 **对您自己的登录数据进行地理定位：** ``` python3 src/geolocate.py --input your_logs.csv --output your_logs_located.csv --ip-column ip_address ``` 该工具在调用 API 之前会对 IP 地址进行去重（因此，如果同一个 IP 在日志中出现了 100 次，也只会触发 1 次 API 调用），并且能够优雅地处理失败情况——私有/保留 IP、无法访问的地址或网络错误都会被记录下来并留空，而不会导致整个运行过程崩溃。 ## 运行测试 ``` python3 -m unittest discover -s tests -v ``` 这会运行以下内容的单元测试： - **`haversine_distance`** - 对照已知的真实世界距离（纽约 ↔ 伦敦）进行验证，以及边缘情况（同一点、对跖点、对称性） - **`flag_impossible_travel`** - 边界条件（刚好处于阈值、缺失数据） - **`compute_risk_score`** - 验证分数保持在 0-100 之内，以及极端与边界情况是否被划入正确的风险等级 - **`geolocate_ip` / `geolocate_csv`** - 使用模拟的 API 响应（测试中没有真实的网络调用），涵盖成功查找、私有 IP、网络错误和 IP 去重 ## 工作原理 1. 每个登录事件都有一个用户、时间戳和位置（经纬度）。 2. 对于每个用户，我们将每次登录与该用户的*上一次*登录（按时间顺序排序）进行比较。 3. 我们使用 Haversine 公式计算两个位置之间的真实世界距离（考虑了地球曲率）。 4. 我们计算两次登录之间流逝的时间。 5. 我们计算出隐含速度 = 距离 ÷ 时间。 6. 如果该速度超过了合理的最大值（比商业航班快，默认为 ~900 km/h），该登录将被标记为“不可能的旅行”——这是账户可能已被攻陷的强烈信号。 7. 每个被标记的事件都会获得一个 **0-100 的风险分数**，而不仅仅是一个“是/否”的标志，该分数结合了两个因素： - **隐含速度超过阈值的幅度**（20,000 km/h 远比 950 km/h 更可疑） - **时间窗口的短促程度**（8 分钟远比 3 小时更可疑） 分数被划分入 LOW / MEDIUM / HIGH / CRITICAL，以便您可以快速浏览结果，被标记的事件会按风险分数排序（最高者优先）——这让分析师能够首先对最危险的事件进行分类处理，而不是把每一个标志都视为同等紧急。 ## 项目结构 ``` geosentinel/ ├── src/ │ ├── generate_data.py # creates synthetic test data with seeded anomalies │ ├── make_sample_raw_log.py # creates a small IP-only sample log │ ├── geolocate.py # real IP -> location, via ip-api.com │ ├── detect.py # core detection: haversine, flagging, risk scoring │ └── app.py # Flask backend serving the dashboard + JSON API ├── templates/ │ └── index.html # dashboard page ├── static/ │ ├── css/style.css # dashboard styling │ ├── js/dashboard.js # dashboard frontend logic │ └── vendor/leaflet/ # Leaflet.js, vendored locally (no CDN dependency) ├── tests/ │ ├── test_detect.py │ └── test_geolocate.py ├── data/ # generated/sample CSVs live here ├── requirements.txt └── README.md ``` ## 可能的后续步骤 本项目涵盖了端到端的核心技术，但真实的系统会走得更远。关于扩展它的想法： - **ASN/VPN 检测**：标记来自已知 VPN/数据中心 IP 范围的登录，这既是攻击的常见来源，也是误报的常见来源 - **地理围栏**：标记来自用户以前从未使用过的国家的登录，无论其速度如何 - **警报**：在检测到 CRITICAL 事件时发送 Slack/电子邮件通知，而不仅仅是记录日志 - **持久化存储**：将基于 CSV 的 pipeline 替换为真实的数据库，以便仪表盘能够显示随时间变化的历史趋势

标签：安全运营, 异常检测, 扫描框架, 逆向工具