stsobhani/siem-platform

# SIEM 平台：企业级安全信息与事件管理 [](https://github.com/stsobhani/siem-platform/actions/workflows/tests.yml) 一个从零开始使用 Python 构建的全栈、作品集级别的 SIEM：它从四种 不同的来源采集日志，将其标准化为统一的 schema，运行六条基于规则的检测以及 无监督的 ML 异常模型，在实时的 SOC 仪表板上将所有内容可视化，并将 相同的数据转发到 **Splunk** 进行企业级搜索和仪表板展示。 旨在通过日志解析/标准化、检测工程、威胁 情报关联、异常检测和 SIEM 工具（自定义构建 *以及* Splunk）展示安全分析的实操技能。 **数据流：** 原始日志行 → 特定来源解析器 → 标准化事件（时间戳、用户名、 源/目标 IP、事件类型、严重程度） → PostgreSQL → 检测规则 + ML 评分 → 告警 → Streamlit 仪表板 *以及* Splunk（通过 HTTP Event Collector）。 ## 功能 ### 1. 多源日志采集 | 来源 | 模拟格式 | 提取的事件类型示例 | |---|---|---| | Windows 安全事件日志 | `Get-WinEvent` / WEF 文本导出 | 登录成功/失败、权限提升 (4672)、组更改 (4728) | | Linux SSH 和 sudo | `/var/log/auth.log` syslog | 登录成功/失败、sudo-to-root 权限提升 | | 防火墙 | pfSense/OPNsense CSV 导出 | 允许/阻止的连接 | | Web 服务器 | Apache/nginx 组合日志格式 | HTTP 请求、认证失败 (401/403)、可疑路径探测 | 每个解析器都实现相同的接口，并输出相同的标准化 schema （`timestamp`、`username`、`src_ip`、`dst_ip`、`event_type`、`severity` 等）。详见 [`siem/parsers/`](siem/parsers/)。 ### 2. PostgreSQL 存储 五张表：`raw_logs`（审计跟踪）、`normalized_events`、`alerts`、`threat_intel_ips`、 `ml_anomaly_scores`。Schema 在 [`sql/init.sql`](sql/init.sql) 中，ORM 模型在 [`siem/db/models.py`](siem/db/models.py) 中。 ### 3. 检测规则引擎：6 条规则 | 规则 | 捕获内容 | |---|---| | **暴力破解登录** | 一个源 IP 在 5 分钟窗口内出现 ≥5 次失败登录 | | **不可能的轨迹** | 同一用户在比物理上可能的速度更快的时间内，从两个地理位置相距甚远的地方成功登录 | | **可疑 IP** | 任何源 IP 与威胁情报指标匹配的事件 | | **权限提升** | sudo-to-root、Windows 4672（管理员 token）/ 4728（组成员身份更改） | | **异常登录时间** | 在可配置的工作时间窗口之外成功登录 | | **多次认证失败尝试** | 一个用户在 60 分钟内跨 *任何* 来源出现 ≥10 次失败的认证事件（可捕获基于来源的暴力破解检测容易漏掉的缓慢/低频、跨向量凭证填充攻击） | 在 [`siem/detection/rules.py`](siem/detection/rules.py) 中实现，每条规则都在 [`tests/test_detection_rules.py`](tests/test_detection_rules.py) 中针对合成数据进行了隔离单元测试 （无需 DB）。 ### 4. 机器学习：Isolation Forest 异常检测 每次登录都会被转换为行为特征向量（一天中的小时、星期几、该用户的源 IP 是否为新 IP、过去一小时的失败尝试、24 小时内的登录速度、非工作时间标志），并使用无监督的 Isolation Forest 进行评分，无需标记的攻击数据。详见 [`siem/ml/anomaly_detection.py`](siem/ml/anomaly_detection.py)。 ### 5. Streamlit SOC 仪表板 实时仪表板显示 KPI 标题、随时间变化的告警趋势、严重程度细分、主要攻击 IP、按规则分类的告警、每用户风险评分、登录时间热力图以及 ML 异常散点 图。详见 [`siem/dashboard/app.py`](siem/dashboard/app.py)。 ### 6. 实时数据模拟 [`scripts/live_log_simulator.py`](scripts/live_log_simulator.py) 使用真实的当前时间连续生成 事件，将它们直接写入数据库，并定期重新运行 检测引擎和 ML pipeline，因此仪表板会像真实的 SOC 实时数据流那样更新，而不是显示单一的静态快照。它将普通的背景活动与周期性的暴力破解爆发、威胁情报命中、权限提升、非工作时间登录以及不可能的轨迹场景混合在一起。检测和 ML 结果在重新运行时会进行去重，因此只有出现真正的新证据时告警计数才会增加，而不是在每次扫描旧事件时都增加。 ### 7. Splunk 集成 - [`scripts/send_to_splunk.py`](scripts/send_to_splunk.py) 通过 HTTP Event Collector (HEC) API 将标准化的事件和 告警转发到 Splunk。 - [`splunk/searches.spl`](splunk/searches.spl)：10 个可直接运行的 SPL 查询（暴力破解 检测、顶级攻击者、权限提升、风险排行榜、多向量关联等）。 - [`splunk/dashboards/siem_overview.xml`](splunk/dashboards/siem_overview.xml)：一个 Splunk Simple XML 仪表板（KPI、按严重程度划分的时间图表、主要攻击 IP、按用户划分的风险表）。 ### 8. Docker、测试、文档 - `docker-compose.yml` 启动 Postgres、采集/检测/ML pipeline、仪表板， 以及（可选）本地 Splunk 实例，适用于确实安装了 Docker 的用户。 - 31 个单元测试（`pytest`），覆盖每个解析器、每条检测规则和 ML pipeline。 ## 快速开始 (Windows, 无需 Docker) ``` python -m venv venv .\venv\Scripts\Activate.ps1 pip install -r requirements.txt # 首先在 PostgreSQL 中创建 siem_user / siem 数据库（如果尚未创建，请参见下文）， # 然后它们将与项目默认值匹配，因此不需要环境变量。 python scripts\generate_sample_logs.py python -m siem.ingestion.pipeline --all --init-db python -m siem.detection.run python -m siem.ml.run streamlit run siem\dashboard\app.py ``` 如果你还没有创建数据库，请以 `postgres` 超级用户身份打开 `psql` 并运行： ``` CREATE USER siem_user WITH PASSWORD 'siem_password'; CREATE DATABASE siem OWNER siem_user; GRANT ALL PRIVILEGES ON DATABASE siem TO siem_user; ``` ## 快速开始 (Docker, 如果可用) ``` git clone https://github.com/stsobhani/siem-platform.git cd siem-platform docker compose up --build ``` 这将启动 PostgreSQL，生成示例日志，采集并对其进行标准化，运行检测和 ML，并在 `http://localhost:8501` 启动仪表板。 ## 运行实时模拟器 上面的命令仅加载一次固定批次的示例日志。要查看仪表板像实时 SOC 数据流那样更新，请在一个终端中保持仪表板运行，并在第二个终端中启动模拟器： ``` .\venv\Scripts\Activate.ps1 python scripts\live_log_simulator.py ``` 它会一直运行直到使用 Ctrl+C 停止，每隔几秒写入一批新事件，并 定期重新运行检测和 ML 评分。在仪表板侧边栏中点击“刷新数据” （或重新加载页面）以获取最新结果。实用标志： ``` python scripts\live_log_simulator.py --interval 5 --duration 600 --detection-every 3 --ml-every 6 ``` `--interval` 是事件批次之间的秒数，`--duration` 是运行的总秒数（0 表示 一直运行直到停止），而 `--detection-every` / `--ml-every` 控制在检测和 ML 运行之间经过多少个 tick。 ## 在本地设置 Splunk (无需 Docker) Splunk 提供 Splunk Enterprise 的免费试用版，可直接在 Windows 上安装： 1. 从 [splunk.com/download](https://www.splunk.com/en_us/download/splunk-enterprise.html) 下载 Windows 安装程序。 2. 运行安装程序，出现提示时设置管理员密码，并保留默认端口 (8000)。 3. 安装完成后，Splunk Web 将在 `http://localhost:8000` 打开。使用你创建的 管理员账户登录。 4. 启用 HTTP Event Collector：依次点击 Settings、Data Inputs、HTTP Event Collector，然后点击 New Token。将其命名为 `siem-token`，保留默认的 source type，并在出现提示时创建一个名为 `siem` 的新 index。复制最后显示的 token 值。 5. 回到 PowerShell，激活你的虚拟环境： $env:SPLUNK_HEC_URL = "https://localhost:8088/services/collector" $env:SPLUNK_HEC_TOKEN = "" $env:SPLUNK_INDEX = "siem" python scripts\send_to_splunk.py --what all 6. 在 Splunk Web 中，转到 Search and Reporting 并运行 `index=siem` 以确认事件已到达。 将 [`splunk/dashboards/siem_overview.xml`](splunk/dashboards/siem_overview.xml) 作为新的仪表板导入 （依次点击 Settings、User Interface、Views、Create New View，然后切换到 Source 并粘贴 XML），并尝试在 Search 中运行 [`splunk/searches.spl`](splunk/searches.spl) 中的查询。 7. 运行 `python scripts\live_log_simulator.py`，同时配合计划的或重复的 `send_to_splunk.py` 调用，以保持 Splunk 中填充最新数据，方便截图或演示。 这为你提供了一个真实运行的 Splunk 实例，包含你自己编写并运行的真实仪表板和真实的 SPL 查询，值得在面试中准确描述。 ## 部署为实时网页 要将其作为链接共享，而不是仅在你的机器上运行的东西，你需要一个可从互联网访问的数据库（你 PC 上的 Postgres 实例无法访问）以及一个托管 Streamlit 应用的地方。以下两者均提供免费层级： 1. **托管 PostgreSQL。** 在 [Neon](https://neon.tech) 上创建一个免费数据库。这将为你提供一个连接字符串，请从中记下主机、端口、 数据库名称、用户名和密码。 2. **将代码推送到 GitHub。** 提交除 `.env` 以及 `.gitignore` 中列出的任何内容之外的所有内容。 3. **在 Streamlit Community Cloud 上部署。** 转到 [share.streamlit.io](https://share.streamlit.io)，使用 GitHub 登录，选择此 repo，并将 主文件路径设置为 `siem/dashboard/app.py`。 4. **将你的数据库凭证添加为 secrets**，而不是代码中的环境变量：在 Streamlit Community Cloud 应用设置中，打开 Secrets，并添加： PG_HOST = "your-neon-or-supabase-host" PG_PORT = "5432" PG_DB = "your-database-name" PG_USER = "your-username" PG_PASSWORD = "your-password" Streamlit 会自动将这些作为环境变量加载，`siem/config.py` 已经 读取了这些变量。 5. **将数据从你自己的机器加载到托管数据库一次**，指向托管 凭证而不是 localhost： $env:PG_HOST = "your-neon-or-supabase-host" $env:PG_USER = "your-username" $env:PG_PASSWORD = "your-password" $env:PG_DB = "your-database-name" python scripts\generate_sample_logs.py python -m siem.ingestion.pipeline --all --init-db python -m siem.detection.run python -m siem.ml.run 6. 你的应用将在 `https://stsobhani-siem-platform-siemdashboardapp-avehhr.streamlit.app/` 实时运行。 ## 运行测试 ``` pytest tests/ -v ``` ## 验证 该项目在开发过程中针对真实的 PostgreSQL 实例进行了端到端的构建和验证， 而不仅仅是编写完毕就假定其可以正常工作。当前状态： | 检查 | 状态 | 如何验证 | |---|---|---| | 单元测试（共 31 个：解析器、检测规则、ML pipeline） | 通过 | 通过上述 GitHub Actions 工作流在每次推送时自动运行。在本地使用 `pytest tests/ -v` 运行 | | 日志解析 | 190/190 行示例日志标准化成功 | `python -m siem.ingestion.pipeline --all --init-db`，对照总行数检查报告的标准化计数 | | 检测引擎 | 全部 6 条规则针对内置的攻击场景正确触发 | `python -m siem.detection.run`，然后检查 `alerts` 表 | | ML 异常检测 | 在没有标记训练数据的情况下，正确地将合成的“不可能的轨迹”和非工作时间登录标记为最高异常 | `python -m siem.ml.run`，然后检查 `ml_anomaly_scores` | | 仪表板 | 启动无错误并通过 HTTP 提供服务 | `streamlit run siem/dashboard/app.py` | 单元测试由 CI 自动执行。Pipeline 级别的检查（解析率、 哪些规则触发、异常评分）鉴于内置的示例数据和日志生成器中固定的 随机种子是确定性的，因此在干净的数据库上重新运行上述命令应该 重现相同的结果。 ## 技术栈 `Python`、`PostgreSQL`、`SQLAlchemy`、`pandas`、`scikit-learn`（Isolation Forest）、 `Streamlit`、`Plotly`、`Docker` / `docker-compose`、`Splunk`（HEC、SPL、仪表板）、 `pytest` ## Repo 结构 ``` siem-platform/ ├── siem/ │ ├── parsers/ # one parser per log source, common NormalizedEvent output │ ├── ingestion/ # raw to normalized to Postgres pipeline │ ├── detection/ # 6 detection rules, threat intel, geoip │ ├── ml/ # Isolation Forest anomaly detection │ ├── dashboard/ # Streamlit SOC dashboard │ └── db/ # SQLAlchemy models and session/engine ├── scripts/ │ ├── generate_sample_logs.py # synthetic logs with embedded attack scenarios │ ├── live_log_simulator.py # continuous live data feed for demos │ └── send_to_splunk.py # HEC forwarder ├── splunk/ │ ├── searches.spl # sample SPL queries │ └── dashboards/siem_overview.xml ├── sql/init.sql # schema ├── tests/ # 31 unit tests (parsers, rules, ML) ├── .streamlit/config.toml ├── docker-compose.yml ├── Dockerfile / Dockerfile.dashboard ``` ## 关于真实性的说明 / 模拟内容 这是一个作品集项目，旨在无需任何外部 API 密钥或付费 服务即可在任何地方运行，因此有几部分内容被有意简化了，代码中指出了清晰的升级路径： - **GeoIP**（[`siem/detection/geoip.py`](siem/detection/geoip.py)）使用小型的静态 IP 段表而不是 MaxMind GeoLite2 数据库，因此“不可能的轨迹”检测可以 离线工作。替换为真实的 GeoIP 数据库只需更改一个函数（`locate(ip)`）。 - **威胁情报**（[`siem/detection/threat_intel.py`](siem/detection/threat_intel.py)）提供了 一个小型的内置 CSV 格式示例指标，而不是实时的 AbuseIPDB/OTX 数据流。该规则 本身永远只从 `threat_intel_ips` 表中读取数据，因此将其指向真实的数据流只是 加载器的替换，而不涉及规则逻辑的更改。 - **示例日志**是合成的（[`scripts/generate_sample_logs.py`](scripts/generate_sample_logs.py)）， 故意嵌入了攻击场景，以便每条规则和 ML 模型都能立即找到 目标，但解析器本身是根据真实的日志格式 （Windows 安全事件字段、实际的 `auth.log` syslog 语法、pfSense CSV 导出、Apache 组合日志格式）构建的，并且无需修改即可解析这些格式的真实日志。

标签：Kubernetes, Python, Streamlit, 安全运营, 异常检测, 扫描框架, 无后门, 测试用例, 访问控制, 请求拦截, 逆向工具