sushiomsky/taso

# TASO – Telegram 自主安全操作员 [](https://python.org) [](LICENSE) [](https://docker.com) ## 概述 TASO 是一个模块化、开源的自主 AI 操作员，它在本地运行并通过私有 Telegram bot 公开其能力。它将多智能体系统、防御性安全工具、威胁情报收集、持久化内存和沙箱化自我改进引擎整合到一个单一、连贯的平台中。 ``` Telegram Interface │ Command Gateway (TelegramBot) │ Agent Orchestrator (CoordinatorAgent) │ Multi-Agent System ├── SecurityAnalysisAgent – static analysis, vulnerability detection ├── ResearchAgent – CVE feeds, CISA KEV, Tor intel ├── DevAgent – code review, patch proposals ├── MemoryAgent – vector store, knowledge DB └── SystemAgent – host metrics, log access │ Tool Execution Layer ├── repo_analyzer – LOC, languages, commits, TODOs ├── dependency_scanner – pip-audit / npm audit ├── web_crawler – HTTP + Tor SOCKS5 ├── system_monitor – psutil metrics ├── sandbox_runner – isolated Docker execution ├── git_manager – clone, diff, apply patches, commit └── log_analyzer – structured log search │ Sandbox (Docker) │ Memory + Knowledge System ├── FAISS vector store – semantic search ├── SQLite knowledge DB – CVEs, findings, audit log └── Conversation store – per-chat history │ Self-Improvement Engine ├── CodeAnalyzer – static analysis, complexity, secrets ├── PatchGenerator – LLM-assisted fix proposals └── AutoDeployer – multi-gate safety pipeline ``` ## 功能特性 | 类别 | 能力 | |------------------------|------------| | 🤖 **AI Agents** | 6 个专家智能体，异步消息总线，协调器编排 | | 🛡️ **Security** | Bandit SAST，秘密扫描，依赖项 CVE 审计，通过 LLM 进行代码审计 | | 🌐 **Threat Intel** | NVD REST API v2，CISA KEV 目录，可选 Tor SOCKS5 爬取 | | 🧠 **Memory** | FAISS 语义搜索，SQLite 结构化存储，对话历史 | | 🔧 **Self-Improvement**| 5 道安全关卡 – 测试，静态分析，补丁大小限制 | | 📦 **Sandbox** | Docker 隔离，内存/CPU 限制，无网络，自动清理 | | 📋 **Audit Log** | 记录所有操作，包含执行者、目标、状态、时间戳 | | 🔒 **Access Control** | 仅限管理员的 Telegram 命令，速率限制 | ## 快速开始 ### 前置条件 | 工具 | 版本 | 必需 | |------|---------|----------| | Python | 3.11+ | ✅ | | Docker | 24+ | ✅ (沙箱) | | Git | 2.x | ✅ | | Ollama | latest | 可选 (本地 LLM) | | Tor | 0.4.x | 可选 (威胁情报) | ### 1. 克隆仓库 ``` git clone https://github.com/yourorg/taso.git cd taso/telegram_autonomous_security_operator ``` ### 2. 安装依赖 ``` bash install.sh ``` 或手动安装： ``` python3.11 -m venv .venv source .venv/bin/activate pip install -r requirements.txt ``` ### 3. 配置 ``` cp .env.example .env # 编辑 .env – 必填字段： # TELEGRAM_BOT_TOKEN – 来自 @BotFather # TELEGRAM_ADMIN_IDS – 你的 Telegram user ID(s) # LLM_BACKEND – ollama | openai | anthropic nano .env ``` ### 4. (可选) 启动 Ollama ``` ollama pull llama3 ollama serve ``` ### 5. 运行 TASO ``` source .venv/bin/activate python main.py ``` ### 6. Docker Compose (生产环境推荐) ``` cp .env.example .env && nano .env docker compose up -d # 拉取默认 LLM 模型到 Ollama docker exec taso_ollama ollama pull llama3 ``` ## Telegram 命令 除非另有说明，所有命令都需要管理员身份验证。 | 命令 | 描述 | 仅限管理员 | |---------|-------------|-----------| | `/start` | 欢迎信息并显示角色 | 否 | | `/help` | 列出所有命令 | 否 | | `/tools` | 列出可用工具 | 否 | | `/status` | 系统指标快照 | ✅ | | `/agents` | 最近的智能体任务历史 | ✅ | | `/memory ` | 语义 + CVE 知识搜索 | ✅ | | `/scan_repo [path]` | 仓库静态分析 | ✅ | | `/security_scan [path]` | 全面安全审计 (SAST + 依赖项 + 秘密) | ✅ | | `/code_audit` | 审计代码片段 (命令后粘贴) | ✅ | | `/threat_intel [keywords]` | 从 NVD + CISA 收集 CVE | ✅ | | `/update_self` | 提议自我改进补丁 | ✅ | | `/logs [category]` | 查看最近的日志行 | ✅ | | `/system` | 主机资源指标 | ✅ | **自由文本消息** 被路由到 LLM 以获取具有每个聊天历史的对话响应。 ## 架构详情 ### 消息总线 所有智能体通过异步发布/订阅消息总线进行通信。 ``` Publisher Bus Subscriber │ │ │ │──── BusMessage ──────────>│──── topic match ─────────>│ │ topic: "security.scan_repo" │ │ payload: {...} │ │ reply_to: "bot.reply.123" │ │<──────────────────────────│<──── result ──────────────│ ``` 主题遵循 `domain.action` 约定： | 前缀 | 拥有者 | |--------|-------| | `coordinator.*` | CoordinatorAgent | | `security.*` | SecurityAnalysisAgent | | `research.*` | ResearchAgent | | `dev.*` | DevAgent | | `memory.*` | MemoryAgent | | `system.*` | SystemAgent | ### 内存架构 ``` User query / agent findings │ ▼ VectorStore (FAISS) ◄── semantic similarity search │ KnowledgeDB (SQLite) ◄── structured: CVEs, analyses, audit log │ ConversationStore (SQLite) ◄── per-chat LLM history ``` ### 自我改进安全关卡 ``` Patch proposal │ Gate 1 ─ Protected module check (config/, sandbox/, self_improvement/) │ Gate 2 ─ Patch size limit (< MAX_PATCH_LINES lines) │ Gate 3 ─ git apply --check (syntactically valid diff) │ Gate 4 ─ Test suite in sandbox (all tests must pass) │ Gate 5 ─ Static analysis score (must not regress) │ ✅ Commit + Audit Log ``` ## 配置参考 查看 `.env.example` 了解所有选项。关键设置： | 变量 | 默认值 | 描述 | |----------|---------|-------------| | `TELEGRAM_BOT_TOKEN` | – | 必需：来自 @BotFather 的 bot token | | `TELEGRAM_ADMIN_IDS` | – | 逗号分隔的管理员用户 ID | | `LLM_BACKEND` | `ollama` | `ollama` / `openai` / `anthropic` | | `OLLAMA_MODEL` | `llama3` | Ollama 的模型名称 | | `SELF_IMPROVE_ENABLED` | `false` | 启用自主补丁 | | `MAX_PATCH_LINES` | `500` | 每次自动补丁的最大行数 | | `PROTECTED_MODULES` | `config,sandbox,self_improvement` | 永不自动补丁 | | `TOR_ENABLED` | `false` | 启用 Tor SOCKS5 进行爬取 | | `DOCKER_MEM_LIMIT` | `256m` | 沙箱容器内存 | | `DOCKER_TIMEOUT` | `60` | 沙箱执行超时 (秒) | ## 添加自定义工具 1. 创建 `tools/my_tool.py` 2. 继承 `BaseTool` 并设置 `name`、`description`、`schema` 3. 实现 `async execute(**kwargs) -> Any` 4. `ToolRegistry` 在启动时自动发现它 ``` # tools/my_tool.py from tools.base_tool import BaseTool, ToolSchema class MyCustomTool(BaseTool): name = "my_tool" description = "Does something useful." schema = ToolSchema({ "target": {"type": "str", "required": True, "description": "..."}, }) async def execute(self, target: str, **_): # your logic here return {"result": f"Processed: {target}"} ``` ## 添加自定义智能体 1. 创建 `agents/my_agent.py` 2. 继承 `BaseAgent` 3. 使用总线主题处理程序实现 `_register_subscriptions()` 4. 在 `orchestrator.py` 中实例化并启动 ## 安全注意事项 - Telegram bot 对所有敏感操作强制执行 **仅限管理员访问**。 - 所有代码执行都在 **Docker 沙箱中**进行：无网络、内存限制、CPU 配额、只读文件系统、丢弃权限。 - 自我改进引擎具有 **多重安全关卡** 和全面的审计跟踪。受保护模块永不自动修改。 - 秘密从 `.env` 加载 – 永不硬编码。 - Tor 默认 **禁用**；需显式使用 `TOR_ENABLED=true` 启用。 ## 项目结构 ``` telegram_autonomous_security_operator/ ├── main.py ← entry point ├── orchestrator.py ← lifecycle manager ├── requirements.txt ├── Dockerfile ├── docker-compose.yml ├── install.sh ├── .env.example │ ├── bot/ │ └── telegram_bot.py ← Telegram interface │ ├── agents/ │ ├── message_bus.py ← async pub/sub bus │ ├── base_agent.py ← abstract base + LLM helpers │ ├── coordinator_agent.py ← task routing, state tracking │ ├── security_agent.py ← SAST, secret scan, code audit │ ├── research_agent.py ← NVD, CISA, Tor intel │ ├── dev_agent.py ← code review, patch proposals │ ├── memory_agent.py ← knowledge storage + retrieval │ └── system_agent.py ← host metrics, log access │ ├── tools/ │ ├── base_tool.py ← base class + tool registry │ ├── repo_analyzer.py │ ├── dependency_scanner.py │ ├── web_crawler.py │ ├── system_monitor.py │ ├── sandbox_runner.py │ ├── git_manager.py │ └── log_analyzer.py │ ├── memory/ │ ├── vector_store.py ← FAISS semantic memory │ ├── knowledge_db.py ← SQLite structured storage │ └── conversation_store.py ← per-chat history │ ├── sandbox/ │ ├── docker_runner.py ← low-level Docker execution │ └── test_runner.py ← test suite runner for self-improve │ ├── self_improvement/ │ ├── code_analyzer.py ← AST + pattern-based analysis │ ├── patch_generator.py ← LLM + rule-based patch generation │ └── auto_deployer.py ← multi-gate deployment pipeline │ ├── config/ │ ├── settings.py ← environment-based configuration │ └── logging_config.py ← loguru structured logging │ └── logs/ ← rotating log files (auto-created) ├── combined.log ├── agent.log ├── tool.log ├── security.log ├── self_improvement.log └── error.log ``` ## 智能体工作流示例 ### 工作流 1：仓库安全审计 ``` User → /scan_repo /home/user/myproject → TelegramBot → coordinator.task {command: scan_repo} → CoordinatorAgent → security.scan_repo → SecurityAnalysisAgent ├── bandit SAST analysis ├── regex secret detection └── LLM executive summary → memory.store (findings persisted) → TelegramBot → User: summary + finding counts ``` ### 工作流 2：威胁情报收集 ``` User → /threat_intel log4j → coordinator.task {command: threat_intel, keywords: ["log4j"]} → ResearchAgent ├── NVD API query (keyword: log4j) ├── CISA KEV catalogue └── LLM trend analysis → memory.store_cve (each CVE) → TelegramBot → User: CVE count + analysis ``` ### 工作流 3：对话式安全问答 ``` User → "What's the CVSS score of CVE-2021-44228?" → ConversationStore (add to history) → LLM query with system prompt + history → (optionally) memory.query (vector search for CVE data) → TelegramBot → User: LLM response ``` ## 许可证 MIT License – 详见 [LICENSE](LICENSE) ## 免责声明 TASO 专为 **防御性安全研究** 设计，仅适用于您拥有或明确授权测试的系统。作者不对滥用行为承担责任。所有工具执行均在您的基础设施上本地进行。

标签：AI风险缓解, DevSecOps, Docker, FAISS, Git 版本控制, Petitpotam, PyRIT, Python, Ruby, Telegram 机器人, 上游代理, 人工智能, 多智能体系统, 大语言模型编排, 威胁情报, 安全专业人员, 安全运营, 安全防御评估, 开发者工具, 扫描框架, 无后门, 用户模式Hook绕过, 知识库, 网络安全, 网络安全研究, 自主安全, 自动化运维, 自愈系统, 请求拦截, 逆向工具, 错误基检测, 防御性安全, 隐私保护, 静态代码分析