0xdad0/telegram-credential-harvester
GitHub: 0xdad0/telegram-credential-harvester
用于监控 Telegram 频道、自动提取泄露凭证并通过 Web 界面集中管理的安全研究与威胁情报工具。
Stars: 1 | Forks: 1
# Telegram 凭证收集器
用于监控 Telegram 频道、提取泄露凭证并通过 Web 界面进行管理的自动化系统。专为渗透测试、威胁情报和安全研究而构建。
## 免责声明
**仅适用于：**
- 授权的渗透测试活动
- 安全研究与威胁情报
- 针对自有或客户授权账号的数据泄露验证
- 具有书面授权的红队演练
滥用可能违反隐私和计算机安全法律。用户需对其使用承担全部责任。
## 架构
```
Telegram API
│
▼
TelegramDownloader
├─ scan loop (concurrent channels, incremental via last_message_id)
│ └─ pushes {channel_id, message_id, file_name} → Redis Queue
└─ worker pool (N workers, configurable)
└─ pops tasks → download → parse (Rust/Python) → PostgreSQL
Redis
├─ tch:file_queue — persistent download task queue (FIFO)
├─ tch:processed — set of channel_id:message_id already done
└─ tch:dead_letter — failed tasks (capped 500, retry via API)
Flask web app
├─ channel management
├─ credential search / export
├─ queue monitoring tab
└─ REST API
```
**技术栈：** Python 3.12 · Telethon · Flask · PostgreSQL 15 · Redis 7 · Rust（可选解析器）
## 功能特性
- 持久化 Redis 队列 — 支持重启恢复；正在处理的任务会自动重新排队
- 增量频道扫描 — 仅获取比上次扫描更新的消息（`last_message_id`）
- 并发频道扫描 — 并行扫描所有活跃频道
- 可配置的凭证模式 — 分隔符、URL 前缀窃取器格式、任意附加正则表达式
- Rust 解析器 — 原生速度的凭证提取（启用自定义模式时回退至 Python）
- 域名定位 — 仅保存匹配特定域名的凭证
- 文件大小限制 — 跳过过大的文件（默认限制为 10 GB）
- 消息文本提取 — 可选从消息正文中提取凭证（不仅限于附件）
- `FloodWait` 处理 — 在触发 Telegram 速率限制时自动休眠，并重新排队任务
- 死信队列 — 失败的任务可通过 Web UI 进行检查和重试
- 强制重新下载 — 清空数据库中的 `processed_files` 可在下次启动时触发全面重新扫描
- 批量频道导入 — 通过 POST 提交 JSON 格式的频道列表
- 统计缓存 — 后台预计算，在插入新凭证时使其失效
- Web 界面 — 搜索、导出 CSV、频道管理、队列监控
## 前置条件
- Docker & Docker Compose
- Telegram API 凭证 — https://my.telegram.org/apps
- 活跃的 Telegram 账号
## 安装说明
### 1. 克隆
```
git clone
cd telegram-credential-harvester
```
### 2. 配置
```
cp .env.prod .env
nano .env
```
最低必填字段：
```
TELEGRAM_API_ID=your_api_id
TELEGRAM_API_HASH=your_api_hash
TELEGRAM_PHONE=+1xxxxxxxxxx
POSTGRES_PASSWORD=change_this
```
### 3. 启动
```
docker-compose up -d
```
### 4. 认证
打开 http://localhost:5000 → Authentication（认证）标签页 → Send Verification Code（发送验证码） → 输入在 Telegram 上收到的验证码。
## 配置参考
所有设置均通过环境变量（`.env` 文件或 Docker Compose）进行。
### Telegram
| 变量 | 默认值 | 描述 |
|---|---|---|
| `TELEGRAM_API_ID` | — | 必填。来自 my.telegram.org |
| `TELEGRAM_API_HASH` | — | 必填。来自 my.telegram.org |
| `TELEGRAM_PHONE` | — | 必填。国际格式 |
### 数据库
| 变量 | 默认值 | 描述 |
|---|---|---|
| `POSTGRES_HOST` | `localhost` | PostgreSQL 主机 |
| `POSTGRES_PORT` | `5432` | PostgreSQL 端口 |
| `POSTGRES_DB` | `credentials_db` | 数据库名称 |
| `POSTGRES_USER` | `admin` | 数据库用户 |
| `POSTGRES_PASSWORD` | `changeme` | **请修改此项** |
### Redis
| 变量 | 默认值 | 描述 |
|---|---|---|
| `REDIS_HOST` | `localhost` | Redis 主机（在 Docker 中为 `redis`） |
| `REDIS_PORT` | `6379` | Redis 端口 |
| `REDIS_DB` | `0` | Redis 数据库索引 |
### 扫描
| 变量 | 默认值 | 描述 |
|---|---|---|
| `SCAN_INTERVAL` | `300` | 完整扫描周期之间的间隔秒数 |
| `WORKER_CONCURRENCY` | `2` | 并行下载 worker 数量 |
| `DELETE_AFTER_PROCESSING` | `true` | 解析后删除本地文件 |
| `MAX_FILE_SIZE_MB` | `10240` | 跳过大于此大小的文件（MB）。`0` = 不限制 |
### 凭证提取
| 变量 | 默认值 | 描述 |
|---|---|---|
| `CREDENTIAL_SEPARATORS` | `[":", "\|", ";", "\t", ","]` | 用于 `emailpassword` 模式的分隔符 JSON 数组 |
| `ENABLE_URL_PREFIX_PATTERN` | `false` | 启用窃取器日志格式：`https://url:email:password` |
| `EXTRACT_FROM_MESSAGE_TEXT` | `false` | 同时扫描消息正文文本（不仅限于文件附件） |
| `EXTRA_CREDENTIAL_PATTERNS` | `[]` | 原始 Python 正则表达式字符串的 JSON 数组；组 1 = 邮箱，组 2 = 密码 |
**自定义模式示例：**
```
EXTRA_CREDENTIAL_PATTERNS=["([\\w.+-]+@[\\w-]+\\.[\\w.]+)\\s*=>\\s*(.+)"]
```
### 过滤
| 变量 | 默认值 | 描述 |
|---|---|---|
| `TARGET_DOMAINS` | `` | 逗号分隔的要保存的域名（留空 = 全部）。示例：`gmail.com,company.it` |
### 应用程序
| 变量 | 默认值 | 描述 |
|---|---|---|
| `WEB_PORT` | `5000` | Web 界面端口 |
| `DOWNLOAD_PATH` | `./downloads` | 下载文件的临时目录 |
## 支持的凭证格式
默认模式（均可通过 `CREDENTIAL_SEPARATORS` 配置）：
| 格式 | 示例 |
|---|---|
| `email:password` | `user@example.com:P@ssw0rd` |
| `email\|password` | `user@example.com\|P@ssw0rd` |
| `email;password` | `user@example.com;P@ssw0rd` |
| `emailpassword` | `user@example.com P@ssw0rd` |
| `email,password` | `user@example.com,P@ssw0rd` |
可选模式（通过环境变量启用）：
| 格式 | 环境变量 |
|---|---|
| `https://url:email:password` | `ENABLE_URL_PREFIX_PATTERN=true` |
| `https://url \| email \| password` | `ENABLE_URL_PREFIX_PATTERN=true` |
| 消息正文文本 | `EXTRACT_FROM_MESSAGE_TEXT=true` |
编码回退链：UTF-8 → Latin-1 → CP1252 → ISO-8859-1。
## Web 界面
| 标签页 | 描述 |
|---|---|
| Authentication | Telegram 登录流程 |
| Search | 全文 + 正则表达式 + 域名搜索，批量删除，导出 CSV |
| Channels | 添加/删除/切换频道，从 JSON 批量导入 |
| Discover Channels | 列出该账号可访问的所有 Telegram 对话 |
| Queue | 实时队列监控 — 待处理文件、死信队列、自动刷新 |
| Statistics | 热门域名、各频道计数、数据库大小 |
| Settings | API 配置、数据库限制、批量删除操作 |
### 队列标签页
显示 Redis 处理 pipeline 的实时状态：
- **Files in Queue** — 等待下载的任务
- **Processed** — 自首次运行以来处理的总文件数
- **Dead-letter** — 失败的任务；点击 **Retry All** 重新排队
- **Redis Connection** — 实时连接状态
- 自动刷新开关（每 5 秒一次）
## API 参考
### 频道
| 方法 | Endpoint | 描述 |
|---|---|---|
| `GET` | `/api/channels` | 列出受监控的频道 |
| `POST` | `/api/channels` | 添加频道 `{"channel_identifier": "@name"}` |
| `DELETE` | `/api/channels/` | 移除频道 |
| `POST` | `/api/channels//toggle` | 切换活跃状态 |
| `POST` | `/api/channels//scan` | 手动扫描（将新文件加入队列） |
| `POST` | `/api/channels/bulk` | 批量导入 `[{"identifier": "...", "name": "..."}]` |
### 队列
| 方法 | Endpoint | 描述 |
|---|---|---|
| `GET` | `/api/queue/status` | `{queue_length, processed_count, dead_letter_count, redis_connected}` |
| `GET` | `/api/queue/items?limit=200` | 查看待处理任务（按时间最早的优先） |
| `GET` | `/api/queue/dead-letter?limit=50` | 检查失败的任务 |
| `POST` | `/api/queue/dead-letter/retry` | 将所有死信任务重新排队 |
### 凭证
| 方法 | Endpoint | 描述 |
|---|---|---|
| `GET` | `/api/credentials/search` | 使用 `?q=&domain=®ex=&page=&per_page=` 进行搜索 |
| `GET` | `/api/credentials/export` | 将过滤后的结果导出为 CSV |
| `DELETE` | `/api/credentials/` | 删除单个凭证 |
| `POST` | `/api/credentials/bulk-delete` | 按域名 / 频道 / 日期删除 |
### 统计
| 方法 | Endpoint | 描述 |
|---|---|---|
| `GET` | `/api/stats` | 完整统计信息（使用缓存） |
| `POST` | `/api/stats/refresh` | 强制重新计算 |
| `POST` | `/api/stats/invalidate` | 使缓存失效 |
## 批量频道导入
将通过 OSINT 发现的频道填充到 `channels_seed.json` 中，然后导入：
```
curl -X POST http://localhost:5000/api/channels/bulk \
-H "Content-Type: application/json" \
-d '[{"identifier": "-1001234567890", "name": "Label"}, {"identifier": "@channelname"}]'
```
频道发现来源：tgscan.io, telemetr.io, 威胁情报源（KELA, Hudson Rock, Flashpoint），Telegram 关键词搜索（`combo`, `logs`, `fullbase`, `stealer`, `leaks`）。
## 项目结构
```
telegram-credential-harvester/
├── docker-compose.yml # PostgreSQL + Redis + app
├── docker-compose.dev.yml # Dev stack (no app container)
├── Dockerfile
├── requirements.txt
├── channels_seed.json # Template for bulk channel import
├── .env.prod # Production config template
├── app/
│ ├── config.py # All settings with defaults
│ ├── database.py # SQLAlchemy + connection pool
│ ├── models.py # credentials, processed_files, telegram_channels, stats_cache
│ ├── queue_manager.py # Redis queue + processed-file set + dead-letter
│ ├── parser.py # Rust/Python credential parser, configurable patterns
│ ├── telegram_client.py # Telethon client, scan loop, worker pool, FloodWait
│ ├── web_app.py # Flask app, all API routes
│ ├── stats_manager.py # Stats cache management
│ ├── background_task.py # Background stats refresh (only when cache invalid)
│ ├── logger.py # Colored terminal logging
│ └── templates/
│ └── index.html # Web UI
├── lib/
│ └── credential_parser_rs/ # Rust parser (optional, ~10x faster on defaults)
├── downloads/ # Temp download directory (auto-created)
└── session/ # Telegram session files (auto-created)
```
## Docker 命令
```
# 启动
docker-compose up -d
# 日志
docker-compose logs -f app
docker-compose logs -f
# 停止
docker-compose down
# 代码更改后重新构建
docker-compose up -d --build
# Shell 访问
docker-compose exec app bash
# PostgreSQL 访问
docker-compose exec postgres psql -U admin -d credentials_db
# Redis CLI
docker-compose exec redis redis-cli
```
## 数据库
### Schema
**`credentials`** — 提取出的凭证对；基于 SHA256(`email:password`) 保证唯一性
**`processed_files`** — 每个下载的文件对应一行记录；`channel_id` + `message_id` 的唯一性防止重复处理
**`telegram_channels`** — 受管理的频道；`last_message_id` 驱动增量扫描
**`stats_cache`** — 预计算的 JSON 数据块，在插入新凭证时失效
### 实用查询
```
-- Top domains
SELECT domain, COUNT(*) AS count
FROM credentials
GROUP BY domain ORDER BY count DESC LIMIT 20;
-- Files processed per channel
SELECT channel_id, COUNT(*) AS files, SUM(credentials_found) AS creds
FROM processed_files
GROUP BY channel_id ORDER BY creds DESC;
-- Credentials discovered today
SELECT email, domain, source_channel, discovered_at
FROM credentials
WHERE discovered_at >= NOW() - INTERVAL '24 hours'
ORDER BY discovered_at DESC;
-- Duplicate check
SELECT credential_hash, COUNT(*) FROM credentials GROUP BY credential_hash HAVING COUNT(*) > 1;
```
## 强制重新下载
要从头重新处理所有文件：
```
# 选项 1：清空 processed_files 表（下次启动时触发重新下载）
docker-compose exec postgres psql -U admin -d credentials_db \
-c "TRUNCATE processed_files;"
# 选项 2：同时清空凭据
docker-compose exec postgres psql -U admin -d credentials_db \
-c "TRUNCATE credentials, processed_files RESTART IDENTITY;"
# 然后重启应用 — Redis processed 集合从空 DB 同步
docker-compose restart app
```
在启动时，应用程序会将 `processed_files` 同步到 Redis。如果表为空，Redis 集合将被清空，并且所有频道文件都会重新排队。
## 性能
| 特性 | 详情 |
|---|---|
| Rust 解析器 | 在标准模式下比 Python 快约 10 倍；当默认配置未更改时自动启用 |
| 增量扫描 | 每个频道单独追踪 `last_message_id`；每个周期仅获取新消息 |
| 并发频道 | 使用 `asyncio.gather` 并行扫描所有活跃频道 |
| CPU 卸载 | `parse_file` 和 `calculate_file_hash` 通过 `asyncio.to_thread` 在线程池中运行 |
| 批量 upsert | PostgreSQL `INSERT … ON CONFLICT DO NOTHING` — 每个文件单条语句 |
| 文件内去重 | Python 解析器在插入数据库之前对文件内的内容进行去重 |
| Redis 已处理集合 | 在每次下载前进行 O(1) 成员检查（避免在热路径上查询数据库） |
| 连接池 | `pool_size=10, max_overflow=20`（总共 30 个连接） |
| 统计缓存 | 预计算的 JSON，仅在凭证插入导致失效时才重新计算 |
## 故障排除
### 容器无法启动
```
docker-compose logs app
docker-compose logs postgres
docker-compose logs redis
```
### Telegram 连接错误
- 检查 `.env` 中的 `API_ID` 和 `API_HASH`
- 手机号格式必须包含国家代码：`+1xxxxxxxxxx`
- 删除过期的会话：`rm -rf session/*`，然后执行 `docker-compose restart app`
### Redis 连接错误
```
# 检查 Redis 是否正在运行
docker-compose exec redis redis-cli ping
# 应返回：PONG
# 查看 queue 状态
docker-compose exec redis redis-cli llen tch:file_queue
docker-compose exec redis redis-cli scard tch:processed
docker-compose exec redis redis-cli llen tch:dead_letter
```
### 数据库无法连接
```
docker-compose exec postgres pg_isready -U admin
# 核弹级选项：擦除并重新创建
docker-compose down -v
docker-compose up -d
```
### 文件未被处理
1. 确认该频道在 Channels 标签页中处于**活跃**状态
2. 检查 `/api/queue/status` 或 Queue 标签页中的队列深度
3. 检查 `/api/queue/dead-letter` 中的死信队列以获取错误消息
. 检查日志：`docker-compose logs -f app`
5. 验证文件的 MIME 类型是否为 `text/*` 或 `application/octet-stream`
### FloodWait / 速率限制
自动处理 — 客户端会休眠所需的时间并重新排队任务。无需人工干预；通过日志进行监控。
## 备份与恢复
```
# 备份
docker-compose exec postgres pg_dump -U admin credentials_db \
| gzip > backup_$(date +%Y%m%d_%H%M).sql.gz
# 恢复
gunzip -c backup_20250615_1200.sql.gz \
| docker-compose exec -T postgres psql -U admin credentials_db
```
## 安全建议
1. 更改 `POSTGRES_PASSWORD` — 切勿将其保留为 `changeme`
2. 在未经身份验证的情况下，请勿将 5000 端口暴露在互联网上
3. 使用带有 HTTP Basic Auth 或 mTLS 的反向代理（NGINX）进行远程访问
4. 使用防火墙规则限制 Docker 网络访问
5. 切勿将 `TELEGRAM_API_ID` / `TELEGRAM_API_HASH` 提交到版本控制系统中
6. Redis 默认没有密码保护 — 请绑定到 localhost 或在生产环境中使用 `requirepass`
## 许可证
按“原样”提供，不提供任何担保。请负责任地使用，并遵守当地法律。
标签：ESC4, Flask, OSINT, Python, Redis, Rust, StruQ, Telegram, 凭据收集, 可视化界面, 威胁情报, 开发者工具, 搜索引擎查询, 数据泄露, 无后门, 测试用例, 网络流量审计, 网络测绘, 请求拦截, 逆向工具, 通知系统