GunesArici/Security-Log-Pipeline

# 认证日志异常 Pipeline 一个小型的端到端 pipeline，负责接收认证日志，通过基于规则的安全检测器（暴力破解、不可能的旅行、非工作时间访问、首次国家访问）对其进行检测，并在交互式仪表板中展示结果——同时附带一个评估脚本，根据已知的注入异常对检测器的准确性进行评分。 该项目旨在将数据工程/BI 工作（ETL、SQL、仪表板）与网络安全视角（检测逻辑、误报分析、评估方法）结合起来。完全在本地运行，无需云账户或付费服务。 ## 架构 ``` generate_logs.py ---> auth_logs.csv ---> SQLite (auth_logs) (synthetic data, | + injected v ground truth) detectors.py (4 rules) | v SQLite (flagged_events) / \ v v dashboard/app.py evaluate.py (Streamlit UI) (precision/recall vs. ground truth) ``` ## 为什么选择这些工具 选择每一个组件都是为了确保 **$0 成本，零外部账户**： 使用 SQLite 而不是托管数据库，使用 Streamlit 作为仪表板而不是购买 BI 平台许可证，并使用 GitHub Actions 的免费层级进行 CI。无需 API 密钥，无需云订阅。 ## 设置 ``` git clone cd security-log-pipeline python3 -m venv .venv && source .venv/bin/activate # optional but recommended pip install -r requirements.txt python pipeline/etl.py # generates data, loads it, runs detectors python pipeline/evaluate.py # scores detector accuracy streamlit run dashboard/app.py # opens the dashboard in your browser pytest tests/ -v # runs the unit tests ``` ## 检测规则 | 检测器 | 逻辑 | 严重程度 | |---|---|---| | **暴力破解** | 某个用户在 10 分钟窗口内出现 ≥5 次失败登录 | 高 | | **不可能的旅行** | 两次成功登录隐含的移动速度 >500 mph（大圆距离 / 时间间隔） | 高 | | **非工作时间访问** | 在当地时间上午 7 点至晚上 8 点之外成功登录 | 中 | | **新国家** | 某用户首次从之前未见过的国家成功登录 | 中 | 这些规则被有意设计得简单且易于解释，而不是基于 ML —— 在事件响应的背景下，分析师需要知道*为什么*会触发警报，而不仅仅是模型给出的评分超过了某个阈值。 ## 评估结果 `generate_logs.py` 向原本正常的流量中注入 4 个已知异常，并将其记录为真实标签（`data/scenarios.json`，检测器自身从不读取该文件）。`evaluate.py` 会将检测器的输出与该真实标签进行比对： ``` Injected scenarios detected: 4/4 (100% recall) Flags matching a known scenario: 5/9 (56% precision) ``` **那些“额外”的标记并不是误报——这是相关检测的一个真实特性。** 例如，一个注入的异常（从新国家登录）也会顺理成章地触发“不可能的旅行”规则，因为同一此登录既是新国家登录，*也是*相对于用户上次所在位置的物理上极其不合理的跳跃。此外，一旦用户的“最后已知位置”被设置为一个异常城市，他们*下一次从家里的正常登录*也会被判定为不可能的旅行——因为在检测器看来，该用户刚刚被传送回来了。这是生产环境中基于位置的检测系统的现实行为，也是为什么真正的 SOC 工具会按事件关联并对告警去重，而不是将每一次规则触发视为独立事件的部分原因。值得指出的是：本项目并未尝试实现该关联层——如果我进一步扩展此项目，这将是自然的“下一次迭代”。 ## 可能的扩展 - 告警关联/去重（将标记按事件分组，而不是独立对待每一个标记） - 结合同一用户上同时触发的多个标记的 `severity score`（严重程度评分） - 针对高严重性标记的 Slack/email webhook - 将合成数据生成器替换为真实数据集（例如，公开的认证日志数据集）以测试泛化能力 ## 项目结构 ``` security-log-pipeline/ ├── data/ │ └── generate_logs.py # synthetic log + ground-truth generator ├── pipeline/ │ ├── db.py # SQLite schema + load/save │ ├── detectors.py # the 4 rule-based detectors │ ├── etl.py # orchestrates generate -> load -> detect │ └── evaluate.py # precision/recall vs. ground truth ├── dashboard/ │ └── app.py # Streamlit dashboard ├── tests/ │ └── test_detectors.py # unit tests for each detector └── .github/workflows/ci.yml # runs tests + full pipeline on every push ```

标签：AMSI绕过, Kubernetes, Python, SQLite, Streamlit, 云计算, 威胁检测, 安全日志分析, 数据管道, 无后门, 红队行动, 规则引擎, 访问控制, 软件工程, 逆向工具