Ordinaryhumanbeing0/CTI_Platform

# Aegis 网络威胁情报 (CTI) 平台 一个可用于生产环境的、面向 SOC 级网络威胁情报平台的 Flask 后端。 本文档涵盖了 **Phase 1**（Flask 基础架构 + 数据库）和 **Phase 2**（RSS 接入引擎）。 ## 目录 1. [项目概述](#project-overview) 2. [目录结构](#directory-structure) 3. [安装与配置](#installation--setup) 4. [运行服务器](#running-the-server) 5. [API Endpoints](#api-endpoints) 6. [Phase 2 – RSS 接入引擎](#phase-2--rss-ingestion-engine) 7. [数据库模型](#database-models) ## 项目概述 Aegis CTI 从多个 RSS 源接入网络安全威胁情报，去重后， 并将文章存储在 MySQL 数据库中。后续阶段将添加 CVE 扩展（NVD、CISA KEV、 EPSS）以及一个完全交互式的 SOC 仪表板。 ## 目录结构 ``` cti-platform/ │ ├── app.py # Application factory + dev server entry point ├── config.py # Environment-driven configuration ├── requirements.txt # All Python dependencies ├── .env # Your local credentials (never commit) ├── .env.example # Credentials template │ ├── database/ # Database layer │ ├── db.py # SQLAlchemy + Migrate instances │ └── models.py # ORM models (CVE, Article, Vendor, KEV, EPSS) │ ├── routes/ # Flask Blueprints │ ├── main.py # / and /health │ └── rss.py # /api/rss/update and /api/rss/articles │ ├── modules/ # Business logic (no Flask dependencies) │ ├── utils/ │ │ └── logger.py # Centralised logging factory │ └── rss/ │ ├── feeds.json # RSS feed configuration (edit to add feeds) │ ├── fetcher.py # Downloads raw feed data with retry/timeout │ ├── parser.py # Normalises feedparser entries into dicts │ ├── deduplicator.py # URL/GUID duplicate detection against DB │ ├── storage.py # Batch-inserts articles in one transaction │ ├── pipeline.py # Orchestrates fetch->parse->dedup->store │ └── scheduler.py # Optional APScheduler wrapper (disabled by default) │ ├── migrations/ # Alembic migration scripts (auto-generated) ├── logs/ # Runtime log files (auto-created on first run) ├── templates/ # Jinja2 HTML templates └── static/ # CSS, JS, and image assets ``` ## 安装与配置 ### 1. 前置条件 - Python 3.10+ - MySQL Server 8.0+ ### 2. 创建虚拟环境 ``` python -m venv venv .\venv\Scripts\Activate.ps1 # Windows PowerShell # source venv/bin/activate # macOS / Linux ``` ### 3. 安装依赖 ``` pip install -r requirements.txt ``` ### 4. 配置环境变量 ``` cp .env.example .env ``` 使用你的 MySQL 凭据编辑 `.env`： ``` FLASK_APP=app.py FLASK_ENV=development SECRET_KEY=your-very-secret-key DB_HOST=localhost DB_PORT=3306 DB_NAME=cti_platform DB_USER=root DB_PASSWORD=your_password ``` ### 5. 创建 MySQL 数据库 ``` CREATE DATABASE cti_platform; ``` ### 6. 运行数据库迁移 ``` .\venv\Scripts\flask db init # First time only — creates migrations/ folder .\venv\Scripts\flask db migrate -m "Initial schema" .\venv\Scripts\flask db upgrade # Applies migrations to MySQL ``` ## 运行服务器 ``` .\venv\Scripts\flask run --host=127.0.0.1 --port=5000 ``` ## API Endpoints | 方法 | Endpoint | 描述 | |--------|----------|-------------| | `GET` | `/` | 渲染 CTI 仪表板前端 | | `GET` | `/health` | 数据库连通性健康检查 | | `GET/POST` | `/api/rss/update` | 触发完整的 RSS 接入 pipeline | | `GET` | `/api/rss/articles` | 获取已存储的文章（分页） | ### `GET /health` ``` { "status": "ok", "database": "connected" } ``` ### `GET /api/rss/update` 运行完整的接入 pipeline 并返回摘要： ``` { "status": "success", "feeds_processed": 10, "articles_found": 185, "articles_added": 42, "duplicates_skipped": 143, "errors": 0 } ``` ### `GET /api/rss/articles` 查询参数： | 参数 | 类型 | 默认值 | 描述 | |-----------|------|---------|-------------| | `page` | int | 1 | 页码（从 1 开始） | | `per_page` | int | 20 | 每页条目数（最大 100） | | `source` | string | — | 按来源名称过滤（部分匹配，不区分大小写） | 示例：`GET /api/rss/articles?page=1&per_page=5&source=BleepingComputer` ## Phase 2 – RSS 接入引擎 ### 数据流 ``` feeds.json | v fetcher.py <- Downloads each feed; retries on failure; skips bad feeds | v parser.py <- Extracts: title, url, guid, summary, author, | published_date, tags from every entry v deduplicator.py <- Queries DB; skips article if URL or GUID already exists | v storage.py <- Batch-inserts new articles in a single transaction | v MySQL (articles) <- Indexed by url (unique) and guid ``` ### 如何添加新的 RSS 源 打开 [`modules/rss/feeds.json`](modules/rss/feeds.json) 并添加一条记录： ``` { "name": "My New Feed", "url": "https://example.com/rss.xml", "enabled": true } ``` - 设置 `"enabled": false` 可在不删除源的情况下暂停使用。 - 无需更改代码或重启 —— pipeline 会在每次运行时读取此文件。 ### 触发 RSS 更新 **浏览器 / curl：** ``` curl http://127.0.0.1:5000/api/rss/update ``` **Python：** ``` import requests r = requests.get("http://127.0.0.1:5000/api/rss/update") print(r.json()) ``` ### 启用自动调度程序 调度程序在 `modules/rss/scheduler.py` 中实现，但**默认禁用**。 要启用每 N 小时的自动更新： 1. 安装 APScheduler： pip install apscheduler 2. 在 `app.py` 的 `create_app()` 中添加： from modules.rss.scheduler import start_scheduler from modules.rss.pipeline import run_pipeline def scheduled_job(): with app.app_context(): run_pipeline() start_scheduler(scheduled_job, interval_hours=6) ## 数据库模型 | 表 | 关键字段 | |-------|-----------| | `articles` | `id`, `title`, `source`, `author`, `url` (unique), `guid` (indexed), `summary`, `tags`, `published_date` | | `cves` | `id` (CVE-YYYY-NNNN), `cvss`, `severity`, `vendor`, `product`, `description` | | `kevs` | `id`, `cve_id` (FK → cves), `date_added`, `description` | | `epss_scores` | `id`, `cve_id` (FK → cves), `epss`, `percentile` | | `vendors` | `id`, `name` (unique) | 迁移历史由 Alembic 在 `alembic_version` 表中自动跟踪。

标签：Flask, GPT, Python, 威胁情报, 开发者工具, 无后门, 漏洞管理, 网络安全, 自定义脚本, 逆向工具, 隐私保护