franckferman/CassandraCTI

## 关于 **CassandraCTI** —— 得名于特洛伊的预言家，她被诅咒会预言未来的危险却无人相信 —— 这是一个模块化的异步 CTI pipeline，它的作用恰恰相反：它确保您的威胁情报能够真正送达需要它的人手中。 它最初是作为内部私有工具开发的，现在已经开源，旨在实现可组合性，并能适应任何安全团队的工作流。 **它的功能：** - 按计划轮询威胁情报源（RSS 订阅源、勒索软件追踪器、恶意域名列表） - 使用基于 SHA1 的指纹技术和本地 SQLite 存储对事件进行去重 - 通过可配置的规则路由事件（按数据源、标签或 regex） - 从 Jinja2 模板渲染消息并将其推送到 Teams 或 Discord webhook - 暴露 Prometheus 指标以实现可观测性 ## 功能 | 功能 | 详情 | |---|---| | **模块化数据源** | RSS、Ransomware Live、Red Flag Domains — 持续增加中 | | **模块化传输方式** | Microsoft Teams、Discord — 可扩展 | | **智能去重** | SHA1 事件指纹 + SQLite 投递追踪 | | **灵活的路由** | 按数据源前缀、标签或 regex 匹配 | | **Jinja2 模板** | 完全掌控消息格式 | | **批处理与限流** | 对事件进行分组，遵守速率限制 | | **环境变量替换** | YAML 中的 `${VAR_NAME}` — 文件中不包含任何敏感信息 | | **试运行模式** | 在不发送任何内容的情况下验证 pipeline | | **数据回填** | 将过往事件重新发送到某个传输方式 | | **Prometheus 指标** | `cassandra_cti_events_sent`、`cassandra_cti_fetch_total` | | **支持 Docker** | 精简镜像，通过挂载卷进行配置 | ## 快速开始 ### 前置条件 - Python 3.10 或更高版本 - pip ### 安装 ``` git clone https://github.com/franckferman/CassandraCTI.git cd CassandraCTI pip install -r requirements.txt ``` 或者作为包安装（提供 `cassandra` CLI 入口点）： ``` pip install . ``` ### 初始化 在标准的配置目录中创建默认配置文件（`config.yaml` + `connectors.yaml`）： - **Linux / macOS:** `~/.config/cassandra-cti/` - **Windows:** `%APPDATA%\cassandra-cti\` ``` python -m cassandra_cti.cli init ``` 这将创建： ``` ~/.config/cassandra-cti/ ├── config.yaml ← sources, routes, filters, store, logging, metrics ├── connectors.yaml ← transport definitions (Teams / Discord webhooks) └── templates/ ← custom Jinja2 templates (optional) ``` ### 快速开始 **1. 添加一个 webhook 连接器：** ``` # Microsoft Teams cassandra add-connector \ --id "teams-soc" \ --webhook-url "https://your-tenant.webhook.office.com/..." # Discord # （手动编辑 connectors.yaml — 参见 Transports 部分） ``` **2. 添加威胁情报源：** ``` # RSS feed cassandra add-source rss \ --name "CERT-FR Alertes" \ --url "https://cert.ssi.gouv.fr/alerte/feed/" \ --tags "cert,fr" # Ransomware tracker cassandra add-source ransomware_live # Malicious domain list cassandra add-source redflag ``` **3. 验证您的设置：** ``` cassandra doctor config cassandra doctor connector --id teams-soc ``` **4. 运行：** ``` # One-shot cassandra run # 每 5 分钟循环 cassandra run --loop --interval 300 # Dry run（不发送消息） cassandra run --dry-run ``` ## 配置 CassandraCTI 使用两个 YAML 文件： | 文件 | 用途 | |---|---| | `config.yaml` | 数据源、路由、过滤器、存储、日志、指标 | | `connectors.yaml` | 传输方式定义（webhook URL、格式化） | ### config.yaml ``` schema_version: 1 scheduler: mode: oneshot # "oneshot" or "loop" interval_seconds: 300 sources: rss: enabled: true feeds: - name: "CERT-FR Alertes" url: "https://cert.ssi.gouv.fr/alerte/feed/" tags: ["cert", "fr"] - name: "BleepingComputer" url: "https://www.bleepingcomputer.com/feed/" tags: ["news"] ransomware_live: enabled: true lookback_days: 30 url: "https://data.ransomware.live/posts.json" # optional red_flag_domains: enabled: true base_url: "https://dl.red.flag.domains/daily/" # optional filters: title_regex_deny: # Drop matching events - "(?i)sponsored" - "(?i)webinar" title_regex_allow: [] # If non-empty, ONLY allow these max_items_per_source: 50 # Cap per source per run transports: use: ["teams-soc", "discord-alert"] # Reference connectors by ID routes: - name: "ransomware" include_sources: ["ransomware.live"] transports: ["teams-soc"] template: "templates/ransomware_card.j2" - name: "cert-alerts" include_tags: ["cert"] transports: ["teams-soc", "discord-alert"] - name: "vendor-news" include_tags: ["vendor", "microsoft"] transports: ["teams-soc"] - name: "catch-all" include_sources: ["rss:"] # Matches any RSS source transports: ["discord-alert"] store: sqlite_path: ".cassandra_cti.db" seen_ttl_days: 90 logging: level: "INFO" # DEBUG | INFO | WARNING | ERROR metrics: enabled: true host: "0.0.0.0" port: 9108 ``` ### connectors.yaml 连接器的定义独立于 `config.yaml`，以便对其进行独立的密钥管理。在 schema 验证之前，环境变量会使用 `${VAR_NAME}` 语法进行替换。 ``` connectors: - id: "teams-soc" type: "teams" params: webhook_url: ${MSTEAMS_WEBHOOK_SOC} theme_color: "0078D7" # Blue border emojis: true throttle_ms: 1000 batching: enabled: true max_items: 5 - id: "discord-alert" type: "discord" params: webhook_url: ${DISCORD_WEBHOOK_URL} username: "CassandraCTI" avatar_url: "https://..." emojis: true throttle_ms: 500 ``` ## 数据源 ### RSS 获取并解析 RSS/Atom 订阅源。每个订阅源被标识为 `rss:{name}`。 ``` sources: rss: enabled: true feeds: - name: "Krebs on Security" url: "https://krebsonsecurity.com/feed/" tags: ["news"] ``` **`config.example.yaml` 中提供了 30 多个预配置的订阅源**，按类别整理： | 类别 | 示例 | |---|---| | CERT | CERT-FR Alertes、CERT-FR Avis | | Microsoft | Microsoft Security、Sentinel Blog、MSRC | | 厂商 | Cisco、Trend Micro、Proofpoint、CrowdStrike、Kaspersky、Recorded Future、Google TAG、Palo Alto | | 新闻 | Krebs on Security、BleepingComputer、Dark Reading、Hacker News、Threatpost、SANS ISC | | 技术文章 | Adam Chester (XPN)、Modexp、James Forshaw | ### Ransomware Live 追踪来自 [ransomware.live](https://ransomware.live) 的勒索软件团伙活动。 ``` sources: ransomware_live: enabled: true lookback_days: 30 ``` 每个事件代表一次勒索软件攻击： - **标题：** `{Victim} by {Group}` - **标签：** `["ransomware"]` - **原始字段：** `group_name`、`country`、`activity`、`post_url`、`description` - **Emoji：** 自动检测受害者国家的国旗 (🇫🇷 🇬🇧 🇩🇪 🇺🇸 ...) ### Red Flag Domains 从 [red.flag.domains](https://red.flag.domains) 获取每日恶意域名列表。 ``` sources: red_flag_domains: enabled: true ``` 每次运行生成一个包含以下内容的事件： - **标题：** `Red Flag Domains – {YYYY-MM-DD}` - **摘要：** 完整的域名列表（以换行符分隔） - **标签：** `["domains"]` - **原始字段：** `file`、`count`、`date` ## 传输方式 ### Microsoft Teams 将 Adaptive Cards（MessageCard 格式）发送到传入的 webhook。 ``` connectors: - id: "teams-cert" type: "teams" params: webhook_url: ${MSTEAMS_WEBHOOK_CERT} theme_color: "0078D7" # Hex without # emojis: true throttle_ms: 1000 # Minimum: 1000ms batching: enabled: false max_items: 10 ``` **主题颜色建议：** | 颜色 | Hex | 用例 | |---|---|---| | 蓝色 | `0078D7` | 常规 / Microsoft | | 紫色 | `8E44AD` | 厂商新闻 | | 橙色 | `D83B01` | 常规新闻 | | 绿色 | `107C10` | 勒索软件（绿色代表金钱） | | 红色 | `C0392B` | 恶意域名 | ### Discord 将富 Embeds 发送到 Discord webhook。 ``` connectors: - id: "discord-alert" type: "discord" params: webhook_url: ${DISCORD_WEBHOOK_URL} username: "CassandraCTI" avatar_url: "https://..." emojis: true throttle_ms: 500 batching: enabled: true max_items: 5 ``` **自动强制执行的字符限制：** - 标题：256 个字符（超长截断） - 描述：4000 个字符（超长截断并附带提示） ## 路由 路由匹配传入的事件并将它们发送到一个或多个传输方式。第一个匹配的路由生效。 ### 按数据源前缀匹配 ``` routes: - name: "all-rss" include_sources: ["rss:"] # Matches rss:CERT-FR, rss:Krebs, etc. transports: ["discord-alert"] ``` ### 按精确数据源匹配 ``` routes: - name: "ransomware" include_sources: ["ransomware.live"] transports: ["teams-soc"] ``` ### 按标签匹配 ``` routes: - name: "cert-only" include_tags: ["cert"] transports: ["teams-cert", "discord-alert"] ``` ### 按 regex 匹配（标题或数据源） ``` routes: - name: "critical" include_regex: "(?i)(critical|zero.day|0-day|CVE)" transports: ["teams-soc", "discord-alert"] ``` ## CLI 参考 ### `cassandra init` 初始化默认配置文件。 ``` Options: --config PATH Path to config.yaml --connectors PATH Path to connectors.yaml ``` ### `cassandra run` 执行聚合周期。 ``` Options: --config PATH Path to config.yaml --connectors PATH Path to connectors.yaml --loop Run in loop mode --interval INT Loop interval in seconds (default: 300) --sources TEXT Comma-separated source filter (e.g. "rss:,ransomware.live") --dry-run Log only, do not send --verbose Set log level to DEBUG --since TEXT Only process events after this date (ISO8601 or YYYY-MM-DD) --no-dedupe Skip deduplication check ``` **示例：** ``` # 标准运行 cassandra run # 每 10 分钟循环，verbose cassandra run --loop --interval 600 --verbose # 仅获取 ransomware 和 CERT feeds cassandra run --sources "ransomware.live,rss:CERT-FR Alertes" # 回填过去一周的事件，不进行 dedup cassandra run --since 2025-03-12 --no-dedupe # Dry run 以测试新 config cassandra run --dry-run --verbose ``` ### `cassandra add-source` 向 config.yaml 添加数据源。 ``` cassandra add-source rss \ --name "CERT-FR Alertes" \ --url "https://cert.ssi.gouv.fr/alerte/feed/" \ --tags "cert,fr" cassandra add-source ransomware_live cassandra add-source redflag ``` ### `cassandra import-feeds` 从 CSV 文件批量导入 RSS 订阅源。 ``` Format: Name,URL,Tags Tags are pipe-separated: "cert|fr|alerts" ``` ``` cassandra import-feeds feeds.csv ``` ### `cassandra add-connector` 向 connectors.yaml 添加一个 Teams 连接器。 ``` cassandra add-connector \ --id "teams-soc" \ --webhook-url "https://..." \ --theme-color "0078D7" \ --emojis ``` ### `cassandra routes-add` 添加或更新路由。 ``` cassandra routes-add \ --name "cert-alerts" \ --include-tag "cert" \ --transports "teams-cert,discord-alert" ``` ### `cassandra doctor` 验证配置并测试连通性。 ``` # 验证 YAML 和 schema cassandra doctor config # 通过 connector 发送测试消息 cassandra doctor connector --id teams-soc ``` ### `cassandra list` 显示当前的数据源、路由和连接器。 ``` cassandra list ``` ### `cassandra backfill` 将过往事件重新发送到特定的传输方式。 ``` cassandra backfill --to teams-soc --since 2025-03-01 ``` ### `cassandra db-reset` 删除本地 SQLite 数据库（清除所有去重历史记录）。 ``` cassandra db-reset # prompts for confirmation cassandra db-reset --force # skip confirmation ``` ### `cassandra seen-clear` 选择性清除去重历史记录，而无需重置整个数据库。 ``` # 清除来自特定 feed 的所有事件 cassandra seen-clear --source-prefix "rss:BleepingComputer" # 清除在某个日期之前看到的事件 cassandra seen-clear --before 2025-01-01 # 清除在某个日期之后来自某个 source 的事件 cassandra seen-clear --source-prefix "ransomware.live" --since 2025-03-01 ``` ## 模板 消息使用 Jinja2 渲染。可以为每个路由分配自定义模板。 ### 可用的上下文变量 | 变量 | 类型 | 描述 | |---|---|---| | `title` | `str` | 事件标题 | | `source` | `str` | 数据源 ID（例如 `rss:CERT-FR`） | | `summary` | `str` | 正文内容 | | `url` | `str` 或 `None` | 事件 URL | | `emoji` | `str` | 为此数据源计算出的 emoji | | `events` | `list[Event]` | 当前数据块中的所有事件（批处理） | | `raw` | `dict` | 来自数据源的原始数据（字段因数据源而异） | ### 内置模板 | 模板 | 用例 | |---|---| | `templates/rss_default.j2` | 标准 RSS 事件 | | `templates/discord_default.j2` | 针对 Discord 优化的布局 | | `templates/ransomware_card.j2` | 勒索软件事件（团伙、国家、活动） | | `templates/domains_list.j2` | 带有预览的恶意域名列表 | | `templates/batch_default.j2` | 批量多事件消息 | ### 示例：自定义模板 ``` {# templates/my_cert_card.j2 #} 🚨 **{{ title }}** | Field | Value | |---|---| | **Source** | {{ source }} | | **Date** | {{ raw.published if raw.published else "—" }} | {{ summary | truncate(300) }} [Read more]({{ url }}) ``` 在路由中分配它： ``` routes: - name: "cert-alerts" include_tags: ["cert"] transports: ["teams-cert"] template: "templates/my_cert_card.j2" ``` ## Docker ``` # Dockerfile（已包含） FROM python:3.12-slim WORKDIR /app COPY pyproject.toml /app/ RUN pip install --upgrade pip && pip install . COPY . /app ENV CTI_CONNECTORS=/config/connectors.yaml ENV PYTHONUNBUFFERED=1 CMD ["cassandra", "run", "--config", "/config/config.yaml"] ``` **构建并运行：** ``` docker build -t cassandra-cti . docker run \ -v /path/to/your/config:/config \ -e MSTEAMS_WEBHOOK_SOC="https://..." \ -e DISCORD_WEBHOOK_URL="https://..." \ cassandra-cti ``` **在 Docker 中循环运行：** ``` docker run \ -v /path/to/your/config:/config \ -e MSTEAMS_WEBHOOK_SOC="https://..." \ cassandra-cti \ cassandra run --config /config/config.yaml --loop --interval 300 ``` **使用 Docker Compose：** ``` services: cassandra-cti: build: . volumes: - ./config:/config environment: - MSTEAMS_WEBHOOK_SOC=${MSTEAMS_WEBHOOK_SOC} - DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL} command: cassandra run --config /config/config.yaml --loop --interval 300 restart: unless-stopped ``` ## 指标 当 `metrics.enabled: true` 时，CassandraCTI 会暴露 Prometheus 指标。 ``` Endpoint: http://0.0.0.0:9108/metrics ``` | 指标 | 类型 | 标签 | 描述 | |---|---|---|---| | `cassandra_cti_events_sent` | Counter | `route` | 成功发送的事件 | | `cassandra_cti_fetch_total` | Counter | `source`、`status` | 获取尝试次数 (`ok` / `err`) | **Prometheus 抓取配置：** ``` scrape_configs: - job_name: "cassandra-cti" static_configs: - targets: ["cassandra-cti:9108"] ``` ## 架构 ``` ┌─────────────────────────────────────────────────────────┐ │ cassandra run │ └──────────────────────┬──────────────────────────────────┘ │ ┌─────────────▼─────────────┐ │ Sources │ │ ┌──────────────────────┐ │ │ │ RSS (feedparser) │ │ │ │ Ransomware Live │ │ │ │ Red Flag Domains │ │ │ └──────────────────────┘ │ └─────────────┬─────────────┘ │ Events ┌─────────────▼─────────────┐ │ Filter & Deduplicate │ │ title_regex_deny / allow │ │ max_items_per_source │ │ SQLite event store │ └─────────────┬─────────────┘ │ New events only ┌─────────────▼─────────────┐ │ Router │ │ source / tag / regex │ │ → transport IDs │ └─────────────┬─────────────┘ │ Routed chunks ┌─────────────▼─────────────┐ │ Transports │ │ ┌──────────────────────┐ │ │ │ Teams (MessageCard) │ │ │ │ Discord (Embed) │ │ │ └──────────────────────┘ │ │ Jinja2 templates │ │ Batching + throttling │ │ Retry with backoff │ └───────────────────────────┘ ``` ## 许可证 基于 **GNU Affero General Public License v3.0** 授权。 完整条款请参见 [LICENSE](https://github.com/franckferman/CassandraCTI/blob/stable/LICENSE)。

(返回顶部)

## 联系方式 [](mailto:contact@franckferman.fr) [](https://www.linkedin.com/in/franckferman) [](https://www.twitter.com/franckferman)

(返回顶部)

标签：Python, RSS, 信息分发, 威胁情报, 实时处理, 开发者工具, 数据聚合, 无后门, 网络信息收集, 自定义请求头, 请求拦截, 逆向工具