jotarios/challenge-threat-intelligence

# 威胁情报 API 一个高性能的 FastAPI 后端，用于实时安全仪表盘。使安全分析师能够调查恶意指标（IP、域名、URL、文件哈希），映射威胁行为者关系，并分析历史活动数据。 ## 架构 采用 CQRS 与多语言持久化架构：OpenSearch 用于实时指标查询，PostgreSQL 用于活动分析，Redis 用于缓存和预计算视图。 ``` ┌───────────────────────────┐ │ FastAPI App │ │ (Docker container) │ │ │ │ Routers -> Services │ │ Middleware (correlation │ │ IDs, structured logging) │ └────────────┬──────────────┘ │ ┌──────────────────┼─────────────────┐ │ │ │ ┌─────▼───────┐ ┌──────▼──────┐ ┌──────▼──────┐ │ OpenSearch │ │ Redis │ │ PostgreSQL │ │ (Hot Tier) │ │ (Cache) │ │(Cold/OLAP) │ │ │ │ │ │ │ │ Denormalized│ │ Cache-aside │ │ SQLAlchemy │ │ indicators │ │ + pre-comp. │ │ + Alembic │ └─────────────┘ └─────────────┘ └─────────────┘ Endpoints 1&2 All endpoints Endpoints 3&4 ``` ## 前置条件 - Docker 和 Docker Compose - Python 3.12+ - `pip install ".[dev]"`（用于在本地运行种子脚本、测试和开发工具） ## 快速开始 ``` make up # Start OpenSearch, PostgreSQL, Redis, and FastAPI make seed # Run migrations + load 10K indicators from seed data # 访问 http://localhost:8000/docs 获取交互式 API 文档 ``` ## API 端点 | Method | Path | Description | Data Source | |--------|------|-------------|-------------| | GET | `/api/indicators/{id}` | 指标详情，包含关联的行为者、活动、指标 | Redis -> OpenSearch | | GET | `/api/indicators/search` | 多参数分页搜索（类型、值、行为者、活动、日期） | OpenSearch | | GET | `/api/campaigns/{id}/indicators` | 按天/周分组的活动时间线 | Redis -> PostgreSQL | | GET | `/api/dashboard/summary` | 首页统计数据（24h/7d/30d） | Redis (预计算) | | GET | `/health` | 服务连接检查 | 所有服务 | ## 示例请求 ``` # 获取指标详情 curl http://localhost:8000/api/indicators/550e8400-e29b-41d4-a716-446655440000 # 按类型搜索指标 curl "http://localhost:8000/api/indicators/search?type=ip&limit=5" # 使用多个过滤器搜索 curl "http://localhost:8000/api/indicators/search?type=domain&threat_actor=APT-North&page=1&limit=20" # Campaign 时间线 curl "http://localhost:8000/api/campaigns/camp-456/indicators?group_by=day" # Dashboard 摘要（默认 7d） curl http://localhost:8000/api/dashboard/summary # 过去 24 小时的 Dashboard 摘要 curl "http://localhost:8000/api/dashboard/summary?time_range=24h" # 健康检查 curl http://localhost:8000/health ``` ## Make 目标 | Command | Description | |---------|-------------| | `make up` | 通过 Docker Compose 启动所有服务 | | `make down` | 停止并移除所有服务和卷 | | `make seed` | 运行 Alembic 迁移 + 将种子数据加载到 OpenSearch 和 PostgreSQL | | `make migrate` | 运行 Alembic 迁移至最新版本 | | `make revision msg="..."` | 根据模型变更自动生成新的 Alembic 迁移 | | `make test` | 运行单元测试（20 个测试） | | `make test-integration` | 运行集成测试（需要运行服务 + 种子数据） | | `make lint` | 运行 ruff 代码检查 | | `make format` | 运行 ruff 格式化 + 自动修复代码检查问题 | | `make typecheck` | 在严格模式下运行 mypy | | `make check` | 运行 lint + typecheck + tests | | `make logs` | 跟踪 FastAPI 应用程序日志 | ## 项目结构 ``` src/ app/ main.py # FastAPI app, lifespan, middleware config.py # pydantic-settings configuration db.py # SQLAlchemy models (source of truth for DB schema) middleware.py # Correlation ID + request logging models/ # Pydantic request/response schemas routers/ # API route handlers services/ # OpenSearch, PostgreSQL, Redis, cache clients alembic/ env.py # Alembic environment (imports app.db.Base) versions/ # Auto-generated migrations tests/ conftest.py # Async fixtures, DI overrides, mock services test_indicators.py # Indicator endpoint tests test_campaigns.py # Campaign endpoint tests test_dashboard.py # Dashboard endpoint tests test_health.py # Health check tests test_cache.py # Cache service tests test_integration.py# Integration tests (real Docker services) scripts/ seed.py # ETL: runs migrations, then SQLite -> OpenSearch + PostgreSQL data/ schema.sql # Original SQL schema (reference only, models are in src/app/db.py) threat_intel.db # SQLite seed data (gitignored) docs/ PRD.md # Full product requirements ``` ## 技术栈 - **FastAPI**，支持异步处理程序和 Pydantic 验证 - **SQLAlchemy** 2.0 异步 ORM，搭配 asyncpg 驱动 - **Alembic** 用于数据库迁移（从模型自动生成） - **OpenSearch** 2.x，支持嵌套文档映射 - **PostgreSQL** 16（用于本地开发的 Redshift 模拟） - **Redis** 7，配备异步客户端 - **structlog** 用于带关联 ID 的 JSON 结构化日志 - **ruff** 用于代码检查和格式化 - **mypy** 用于静态类型检查（严格模式） - **Docker Compose** 用于本地编排 - **pytest** + **httpx** 用于异步测试套件 ## 开发 ``` # 安装 dev dependencies pip install ".[dev]" # 格式化代码 make format # 运行所有检查（lint + typecheck + tests） make check # 修改数据库 schema # 1. 在 src/app/db.py 中编辑 models # 2. 生成 migration： make revision msg="add new column to indicators" # 3. 应用： make migrate ```

标签：AV绕过, CQRS, Docker, FastAPI, IOC查询, PostgreSQL, Python, Redis, SQLAlchemy, Web安全, 历史攻击活动, 后端开发, 威胁情报, 威胁行为者, 安全仪表盘, 安全规则引擎, 安全运营, 安全防御评估, 实时安全, 开发者工具, 恶意指标分析, 扫描框架, 搜索引擎查询, 数据可视化支持, 无后门, 架构设计, 测试用例, 蓝队分析, 请求拦截, 逆向工具