AI免杀:基于 Evasion‑SubAgents 的Skills设计与实现

admin 2026-07-15 05:13:50 网络安全文章 来源:ZONE.CI 全球网 0 阅读模式

文章总结: 本文介绍Evasion-SubAgents,一个用于AI免杀实验的元工作流系统。其核心设计包括四层架构(代理、命令、技能、知识库)和三层抽象(Agent/Command/Skill),实现角色边界定义、用户接口解耦与执行流程标准化。知识库采用原子化组件设计,支持可复用、可组合的Loader生成。项目通过知识库管理器实现技术去重与结构化存储,为免杀技能编写提供可审计的框架。 综合评分: 86 文章分类: 其他


cover_image

AI免杀:基于 Evasion‑SubAgents 的Skills设计与实现

Arthur Arthur

C4安全

2026年6月8日 08:53 江苏

在小说阅读器读本章

去阅读

原文链接:https://forum.butian.net/ai_security/132

作者:Arthur


开篇:

Evasion SubAgents 是一个用来做免杀实验的工具箱,通过它来学习免杀skill的编写

项目架构总览

一、四层架构设计:从命令到知识的完整链路

这个项目的本质不是传统意义上的独立程序,而是一套元工作流系统:

evasion-agent-teams/
├── .claude-plugin/
│   └── plugin.json                    # 插件元数据配置
├── agents/                            # 代理层:定义角色边界
│   ├── research-agent.md             # 技术研究代理
│   ├── loadergen-agent.md            # Loader生成代理
│   ├── evasion-agent.md              # 免杀集成代理
│   └── c2-evasion-agent.md           # C2免杀代理(最成熟)
├── skills/                            # 技能层:定义执行流程
│   ├── research/SKILL.md             # 搜索分析技能
│   ├── loader_generate/SKILL.md      # Loader生成技能
│   ├── evasion_integrate/SKILL.md    # 免杀集成技能
│   └── c2_evasion/SKILL.md           # C2免杀技能
├── commands/                          # 命令层:用户入口
│   ├── research.md
│   ├── loader_generate.md
│   ├── evasion_integrate.md
│   └── c2_evasion.md
├── lib/
│   └── knowledge_manager.py          # 知识库管理器(核心代码)
├── knowledge-base/                    # 知识库层:结构化存储
│   ├── evasion_techniques.json       # 免杀技术库
│   ├── loader_techniques.json        # Loader组件库
│   └── scenarios.json                # 已验证场景库
├── hooks/
│   └── langfuse_hook.py              # 可观测性钩子
└── output/                            # 输出目录

二、三层抽象:Agent/Command/Skill的职责分离

2.1 Agent层

Agent层的核心是角色边界和权限控制。以c2-evasion-agent.md为例:

# C2 Evasion Agent

## 角色定位
你是专门负责C2框架源码分析和免杀修改的安全专家。

## 权限范围
允许:
- 直接修改源码文件
- 读取YARA/Sigma规则
- 分析二进制特征
- 执行编译验证

禁止:
- 执行恶意代码
- 破坏性修改
- 未经分析直接修改

## 执行原则
1. 先找规则,再决定改什么
2. 逐条分析,动态生成任务
3. 优先编译选项,后考虑源码重构
4. 每条规则产出独立分析文档

关键设计点:Agent不是提供知识,而是定义人格边界和动作权限。

2.2 Command层

Command层将复杂功能封装成产品化的命令入口。以loader_generate.md为例:

# Command: loader_generate

## 功能描述
从loader知识库生成shellcode loader

## 参数说明
- [count] 生成数量(默认1)
- --shellcode <path> 指定shellcode文件
- --executor <type> 执行方式(callback/createRemoteThread等)
- --complexity <level> 复杂度(simple/medium/complex)
- --language <lang> 编程语言(c/rust/go)

## 输出结构
output/
└── loader_<timestamp>/
&nbsp; &nbsp; ├── source/ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;# 生成的源码
&nbsp; &nbsp; ├── build/ &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;# 编译产物
&nbsp; &nbsp; └── metadata.json &nbsp; &nbsp;&nbsp;# 生成参数记录

## 调用Agent
loadergen-agent

设计优势:用户接口与内部逻辑解耦,命令文档即产品帮助页。

2.3 Skill层

Skill层是真正的执行工艺说明书。以c2_evasion/SKILL.md为例:

# Skill: C2 Evasion

## 执行流程

### Phase 1: 识别C2组件
1. 扫描项目目录结构
2. 识别网络通信模块
3. 标记核心功能文件
输出:c2_components.md

### Phase 2: 搜索检测规则
1. 搜索YARA规则(GitHub/官方仓库)
2. 搜索Sigma规则
3. 搜索网络检测规则
输出:detection_rules/目录

### Phase 3: 逐条规则分析
对每条检测规则:
1. 提取检测特征
2. 定位源码对应位置
3. 分析修改可行性
4. 生成修改建议
输出:rule_analysis/<rule_name>.md

### Phase 4: 二进制资产分析
1. 分析十六进制模式
2. 识别硬编码字符串
3. 查找API调用序列
输出:binary_analysis.md

### Phase 5: 敏感字符串搜索
1. 搜索IOC特征
2. 搜索签名相关
3. 搜索调试特征
输出:sensitive_strings.md

### Phase 6: 修改和验证
1. 根据分析生成修改任务
2. 实施源码修改
3. 编译验证
4. 功能测试
输出:modified_source/ + build_results.md

核心价值:将复杂任务拆解为可追溯、可审计的阶段化流程。

三、知识库设计:从碎片化到结构化

3.1 知识库结构设计

知识库不是存储大段文本,而是将知识原子化、组件化:

免杀技术库

{
"techniques": [
&nbsp; &nbsp; {
"id":&nbsp;"T001",
"name":&nbsp;"API Hashing",
"evasion_type":&nbsp;"api_obfuscation",
"description":&nbsp;"通过哈希值动态解析API,避免导入表暴露敏感函数",
"code_template":&nbsp;"/* C语言API哈希实现模板 */",
"apis": ["LoadLibrary",&nbsp;"GetProcAddress"],
"complexity":&nbsp;"medium",
"references": ["GitHub链接"],
"last_updated":&nbsp;"2025-01-01"
&nbsp; &nbsp; }
&nbsp; ]
}

Evasion类型分类:

  • api_obfuscation

  • API混淆

  • string_obfuscation

  • 字符串混淆

  • memory_evasion

  • 内存规避

  • execution_evasion

  • 执行规避

  • anti_analysis

  • 反分析

  • amsi_etw_bypass

  • AMSI/ETW绕过

  • unhooking

  • 脱钩

Loader组件库

{
"storage_methods": [
&nbsp; &nbsp; {
"id":&nbsp;"sto_001",
"name":&nbsp;"File Embedding",
"description":&nbsp;"将shellcode嵌入资源段",
"code_snippet":&nbsp;"/* 资源段嵌入代码 */"
&nbsp; &nbsp; }
&nbsp; ],
"memory_allocators": [
&nbsp; &nbsp; {
"id":&nbsp;"mem_001",
"name":&nbsp;"VirtualAlloc",
"description":&nbsp;"使用VirtualAlloc分配可执行内存",
"apis": ["VirtualAlloc",&nbsp;"VirtualProtect"]
&nbsp; &nbsp; }
&nbsp; ],
"data_copiers": [
&nbsp; &nbsp; {
"id":&nbsp;"cpy_001",
"name":&nbsp;"RtlMoveMemory",
"description":&nbsp;"使用RtlMoveMemory复制shellcode"
&nbsp; &nbsp; }
&nbsp; ],
"executors": [
&nbsp; &nbsp; {
"id":&nbsp;"exe_001",
"name":&nbsp;"CreateThread",
"description":&nbsp;"使用CreateThread执行shellcode"
&nbsp; &nbsp; }
&nbsp; ]
}

设计精髓:将完整Loader拆解为可组合的原子组件,实现:

  • 可复用性
  • 可筛选性
  • 可组合性
  • 可统计性

3.2 知识治理的核心

lib/knowledge_manager.py是项目中最接近传统程序的部分:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
知识库管理器 - 负责知识的增删改查、去重、组合和导出
"""

import json
import os
from&nbsp;typing import Dict,&nbsp;List, Optional, Any
from&nbsp;difflib import SequenceMatcher
from&nbsp;pathlib import Path

classKnowledgeManager:
&nbsp; &nbsp; """知识库管理器类"""

def__init__(self,&nbsp;base_dir:&nbsp;str&nbsp;= "knowledge-base"):
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; 初始化知识库管理器

Args:
base_dir: 知识库基础目录路径
&nbsp; &nbsp; &nbsp; &nbsp; """
self.base_dir&nbsp;=&nbsp;Path(base_dir)
self.evasion_file&nbsp;=&nbsp;self.base_dir&nbsp;/ "evasion_techniques.json"
self.loader_file&nbsp;=&nbsp;self.base_dir&nbsp;/ "loader_techniques.json"
self.scenarios_file&nbsp;=&nbsp;self.base_dir&nbsp;/ "scenarios.json"

&nbsp; &nbsp; &nbsp; &nbsp; # 初始化知识库文件(如果不存在)
self._initialize_knowledge_bases()

def_initialize_knowledge_bases(self):
&nbsp; &nbsp; &nbsp; &nbsp; """初始化知识库文件,创建默认schema"""
default_schemas&nbsp;=&nbsp;{
self.evasion_file: {"techniques": []},
self.loader_file: {
"storage_methods": [],
"memory_allocators": [],
"data_copiers": [],
"executors": []
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; },
self.scenarios_file: {"scenarios": []}
&nbsp; &nbsp; &nbsp; &nbsp; }

for&nbsp;file_path, schema in default_schemas.items():
if&nbsp;not file_path.exists():
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; file_path.parent.mkdir(parents=True, exist_ok=True)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; with open(file_path,&nbsp;'w', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; json.dump(schema, f, indent=2, ensure_ascii=False)

&nbsp; &nbsp; def add_evasion_technique(self, technique: Dict[str, Any]) ->&nbsp;bool:
"""
&nbsp; &nbsp; &nbsp; &nbsp; 添加免杀技术到知识库

&nbsp; &nbsp; &nbsp; &nbsp; Args:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; technique: 技术字典,包含id, name, evasion_type等字段

&nbsp; &nbsp; &nbsp; &nbsp; Returns:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; bool: 添加成功返回True
&nbsp; &nbsp; &nbsp; &nbsp; """
# 检查是否重复(基于相似度)
ifself._is_duplicate_technique(technique):
print(f"注意: 检测到相似技术: {technique['name']}")
returnFalse

# 读取现有数据
&nbsp; &nbsp; &nbsp; &nbsp; with open(self.evasion_file,&nbsp;'r', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; data = json.load(f)

# 添加新技术
&nbsp; &nbsp; &nbsp; &nbsp; data["techniques"].append(technique)

# 写回文件
&nbsp; &nbsp; &nbsp; &nbsp; with open(self.evasion_file,&nbsp;'w', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; json.dump(data, f, indent=2, ensure_ascii=False)

print(f"已添加技术: {technique['name']}")
returnTrue

&nbsp; &nbsp; def _is_duplicate_technique(self, new_technique: Dict) ->&nbsp;bool:
"""
&nbsp; &nbsp; &nbsp; &nbsp; 检测技术是否重复(多维度相似度检查)

&nbsp; &nbsp; &nbsp; &nbsp; 检查维度:
&nbsp; &nbsp; &nbsp; &nbsp; 1. 名称相似度
&nbsp; &nbsp; &nbsp; &nbsp; 2. 关键词重叠度
&nbsp; &nbsp; &nbsp; &nbsp; 3. API重叠度
&nbsp; &nbsp; &nbsp; &nbsp; 4. 来源重叠

&nbsp; &nbsp; &nbsp; &nbsp; Args:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; new_technique: 新技术字典

&nbsp; &nbsp; &nbsp; &nbsp; Returns:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; bool: 如果重复返回True
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; with open(self.evasion_file,&nbsp;'r', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; data = json.load(f)

for&nbsp;existing in data["techniques"]:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; similarity_score =&nbsp;0

# 1. 名称相似度检查
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; name_similarity = SequenceMatcher(
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; None,
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; new_technique['name'].lower(),
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; existing['name'].lower()
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ).ratio()
if&nbsp;name_similarity >&nbsp;0.9:
returnTrue
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; similarity_score += name_similarity *&nbsp;0.3

# 2. 关键词重叠检查
if'keywords'&nbsp;in new_technique&nbsp;and'keywords'&nbsp;in existing:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; keyword_overlap = len(
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; set(new_technique['keywords']) &
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; set(existing['keywords'])
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ) / max(len(new_technique['keywords']),&nbsp;1)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; similarity_score += keyword_overlap *&nbsp;0.3

# 3. API重叠检查
if'apis'&nbsp;in new_technique&nbsp;and'apis'&nbsp;in existing:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; api_overlap = len(
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; set(new_technique['apis']) &
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; set(existing['apis'])
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ) / max(len(new_technique['apis']),&nbsp;1)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; similarity_score += api_overlap *&nbsp;0.3

# 4. 来源重叠检查
if'source'&nbsp;in new_technique&nbsp;and'source'&nbsp;in existing:
if&nbsp;new_technique['source'] == existing['source']:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; similarity_score +=&nbsp;0.1

# 如果综合相似度超过阈值,视为重复
if&nbsp;similarity_score >&nbsp;0.7:
returnTrue

returnFalse

&nbsp; &nbsp; def get_loader_components(self, component_type: str) ->&nbsp;List[Dict]:
"""
&nbsp; &nbsp; &nbsp; &nbsp; 获取指定类型的Loader组件

&nbsp; &nbsp; &nbsp; &nbsp; Args:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; component_type: 组件类型
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;(storage_methods/memory_allocators/data_copiers/executors)

&nbsp; &nbsp; &nbsp; &nbsp; Returns:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; List[Dict]: 组件列表
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; with open(self.loader_file,&nbsp;'r', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; data = json.load(f)

return&nbsp;data.get(component_type, [])

&nbsp; &nbsp; def generate_random_loader(self, count:&nbsp;int&nbsp;=&nbsp;1) ->&nbsp;List[Dict]:
"""
&nbsp; &nbsp; &nbsp; &nbsp; 随机生成Loader组合

&nbsp; &nbsp; &nbsp; &nbsp; Args:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; count: 生成数量

&nbsp; &nbsp; &nbsp; &nbsp; Returns:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; List[Dict]: Loader配置列表
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; import random

&nbsp; &nbsp; &nbsp; &nbsp; loaders = []
for&nbsp;_ in range(count):
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; loader = {
"storage": random.choice(self.get_loader_components("storage_methods")),
"allocator": random.choice(self.get_loader_components("memory_allocators")),
"copier": random.choice(self.get_loader_components("data_copiers")),
"executor": random.choice(self.get_loader_components("executors"))
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; loaders.append(loader)

return&nbsp;loaders

&nbsp; &nbsp; def record_scenario(self, scenario: Dict[str, Any]):
"""
&nbsp; &nbsp; &nbsp; &nbsp; 记录已验证的场景组合

&nbsp; &nbsp; &nbsp; &nbsp; Args:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; scenario: 场景字典,包含components, result, timestamp等
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; with open(self.scenarios_file,&nbsp;'r', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; data = json.load(f)

&nbsp; &nbsp; &nbsp; &nbsp; data["scenarios"].append(scenario)

&nbsp; &nbsp; &nbsp; &nbsp; with open(self.scenarios_file,&nbsp;'w', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; json.dump(data, f, indent=2, ensure_ascii=False)

&nbsp; &nbsp; def get_stats(self) -> Dict[str, Any]:
"""
&nbsp; &nbsp; &nbsp; &nbsp; 获取知识库统计信息

&nbsp; &nbsp; &nbsp; &nbsp; Returns:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Dict: 统计数据
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; stats = {}

# 统计免杀技术
&nbsp; &nbsp; &nbsp; &nbsp; with open(self.evasion_file,&nbsp;'r', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; data = json.load(f)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; stats["evasion_techniques"] = len(data["techniques"])

# 按类型统计
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; type_count = {}
for&nbsp;tech in data["techniques"]:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; tech_type = tech.get("evasion_type",&nbsp;"unknown")
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; type_count[tech_type] = type_count.get(tech_type,&nbsp;0) +&nbsp;1
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; stats["by_type"] = type_count

# 统计Loader组件
&nbsp; &nbsp; &nbsp; &nbsp; with open(self.loader_file,&nbsp;'r', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; data = json.load(f)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; stats["loader_components"] = {
"storage_methods": len(data.get("storage_methods", [])),
"memory_allocators": len(data.get("memory_allocators", [])),
"data_copiers": len(data.get("data_copiers", [])),
"executors": len(data.get("executors", []))
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }

# 统计场景
&nbsp; &nbsp; &nbsp; &nbsp; with open(self.scenarios_file,&nbsp;'r', encoding='utf-8')&nbsp;as&nbsp;f:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; data = json.load(f)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; stats["verified_scenarios"] = len(data.get("scenarios", []))

return&nbsp;stats

关键功能亮点:

  1. 多维度去重逻辑:不仅检查名称完全匹配,还计算名称相似度、关键词重叠、API重叠、来源重叠
  2. 结构化查询:支持按类型、复杂度、API等维度筛选
  3. 场景记录:记录已验证的组合,避免重复劳动
  4. 统计分析:提供知识库使用情况统计

设计哲学:安全知识库最怕的不是少,而是越积越乱。

四、规则驱动的修改流程

4.1 六阶段处理流程

整个项目最成熟、最像真正工作流系统的部分。

4.2 核心设计原则

## C2 Evasion Agent 核心原则

### 禁止行为
- 在分析前预先建立任务
- 在未看到规则前假设应该改什么
- 凭感觉乱改源码
- 一上来就重构源码

### 推荐做法
1. 先搜索检测规则(YARA/Sigma/Network)
2. 逐条规则分析检测特征
3. 定位源码对应位置
4. 优先考虑编译选项和构建配置
5. 最后才修改源码
6. 每条规则产出独立分析文档

这背后的思维:将高不确定性任务改造成证据驱动、步骤可追溯、输出可审计的流程。

Phase 3示例:逐条规则分析

# 伪代码示例:规则分析流程
def analyze_detection_rule(rule_file: str) -> Dict:
"""
&nbsp; &nbsp; 分析单条检测规则

&nbsp; &nbsp; 步骤:
&nbsp; &nbsp; 1. 解析规则文件(YARA/Sigma)
&nbsp; &nbsp; 2. 提取检测特征(字符串/正则/API序列)
&nbsp; &nbsp; 3. 在源码中搜索匹配位置
&nbsp; &nbsp; 4. 评估修改难度和影响
&nbsp; &nbsp; 5. 生成修改建议

&nbsp; &nbsp; 返回:
&nbsp; &nbsp; {
&nbsp; &nbsp; &nbsp; &nbsp; "rule_name": "CobaltStrike_Beacon",
&nbsp; &nbsp; &nbsp; &nbsp; "detection_features": [
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; {"type": "string", "value": "\\x5c\\x4d\\x5c\\x52"},
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; {"type": "api", "value": "InternetOpenA"}
&nbsp; &nbsp; &nbsp; &nbsp; ],
&nbsp; &nbsp; &nbsp; &nbsp; "source_locations": [
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; {"file": "beacon.c", "line": 123, "feature": "M\\R字符串"}
&nbsp; &nbsp; &nbsp; &nbsp; ],
&nbsp; &nbsp; &nbsp; &nbsp; "modification_suggestions": [
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; {"type": "string_encryption", "priority": "high"},
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; {"type": "api_dynamic_resolve", "priority": "medium"}
&nbsp; &nbsp; &nbsp; &nbsp; ]
&nbsp; &nbsp; }
&nbsp; &nbsp; """
&nbsp; &nbsp; pass

五、可观测性的保障

5.1 Tracing

多代理系统面临的核心挑战:调不通、查不到、复盘不了。

hooks/langfuse_hook.py解决了官方对subagent调用链支持不足的问题:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Langfuse Hook - 为Claude Code多代理系统提供可观测性

功能:
1. 捕获主会话和子代理调用
2. 记录tool_use和tool_result
3. 生成trace上传到Langfuse
4. 支持本地状态管理和失败恢复
"""

import json
import hashlib
import fcntl
from&nbsp;pathlib import Path
from&nbsp;langfuse import Langfuse
from&nbsp;typing import Dict, Any, Optional

classLangfuseHook:
&nbsp; &nbsp; """Langfuse钩子类"""

def__init__(self,&nbsp;project_dir:&nbsp;str&nbsp;= "."):
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; 初始化Langfuse钩子

Args:
project_dir: 项目目录
&nbsp; &nbsp; &nbsp; &nbsp; """
self.project_dir&nbsp;=&nbsp;Path(project_dir)
self.state_file&nbsp;=&nbsp;self.project_dir&nbsp;/ ".langfuse_state.json"
self.lock_file&nbsp;=&nbsp;self.project_dir&nbsp;/ ".langfuse.lock"

&nbsp; &nbsp; &nbsp; &nbsp; # 初始化Langfuse客户端
self.langfuse&nbsp;=&nbsp;Langfuse(
public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
host=os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com")
&nbsp; &nbsp; &nbsp; &nbsp; )

defprocess_transcript(self,&nbsp;transcript_path:&nbsp;str):
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; 处理ClaudeCode对话记录

&nbsp; &nbsp; &nbsp; &nbsp; 步骤:
&nbsp; &nbsp; &nbsp; &nbsp; 1. 读取增量内容(基于上次处理位置)
&nbsp; &nbsp; &nbsp; &nbsp; 2. 识别消息类型:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;- 主会话消息
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;-&nbsp;tool_use事件
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;-&nbsp;tool_result事件
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;-&nbsp;subagent调用事件
&nbsp; &nbsp; &nbsp; &nbsp; 3. 生成trace并上传
&nbsp; &nbsp; &nbsp; &nbsp; 4. 更新本地状态

Args:
transcript_path: 对话记录文件路径
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; # 获取文件锁(防止并发冲突)
withopen(self.lock_file, 'w')&nbsp;aslock:
fcntl.flock(lock.fileno(),&nbsp;fcntl.LOCK_EX)

try:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; # 读取上次状态
last_position&nbsp;=&nbsp;self._read_state()

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; # 读取增量内容
withopen(transcript_path, 'r')&nbsp;asf:
f.seek(last_position)
new_content&nbsp;=&nbsp;f.read()
current_position&nbsp;=&nbsp;f.tell()

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; # 解析新增消息
messages&nbsp;=&nbsp;self._parse_messages(new_content)

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; # 处理每条消息
formsginmessages:
ifmsg['type'] == 'subagent_call':
self._trace_subagent_call(msg)
elifmsg['type'] == 'tool_use':
self._trace_tool_use(msg)
elifmsg['type'] == 'tool_result':
self._trace_tool_result(msg)

&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; # 更新状态
self._update_state(current_position)

finally:
fcntl.flock(lock.fileno(),&nbsp;fcntl.LOCK_UN)

def_trace_subagent_call(self,&nbsp;msg:&nbsp;Dict[str,&nbsp;Any]):
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; 记录子代理调用

Args:
msg: 子代理调用消息
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; # 生成traceID(基于内容哈希)
trace_id&nbsp;=&nbsp;hashlib.sha256(
json.dumps(msg,&nbsp;sort_keys=True).encode()
&nbsp; &nbsp; &nbsp; &nbsp; ).hexdigest()[:16]

&nbsp; &nbsp; &nbsp; &nbsp; # 创建Langfusetrace
trace&nbsp;=&nbsp;self.langfuse.trace(
id=trace_id,
name=f"SubAgent:&nbsp;{msg['agent_name']}",
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; input=msg['prompt'],
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; metadata={
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "agent_type": msg['agent_name'],
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "tools_used": msg.get('tools', []),
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "duration": msg.get('duration_ms')
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }
&nbsp; &nbsp; &nbsp; &nbsp; )

&nbsp; &nbsp; &nbsp; &nbsp; # 记录span(子代理执行)
&nbsp; &nbsp; &nbsp; &nbsp; span = trace.span(
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; name="subagent_execution",
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; input=msg['prompt'],
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; output=msg.get('result')
&nbsp; &nbsp; &nbsp; &nbsp; )

&nbsp; &nbsp; &nbsp; &nbsp; # 结束span
&nbsp; &nbsp; &nbsp; &nbsp; span.end()

&nbsp; &nbsp; def _truncate_large_content(self, content: str, max_length: int = 10000):
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; 截断大内容(防止超出Langfuse限制)

&nbsp; &nbsp; &nbsp; &nbsp; Args:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; content: 原始内容
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; max_length: 最大长度

&nbsp; &nbsp; &nbsp; &nbsp; Returns:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; str: 截断后的内容
"""
&nbsp; &nbsp; &nbsp; &nbsp; if len(content) > max_length:
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return content[:max_length] + f"... (truncated, total {len(content)} chars)"
&nbsp; &nbsp; &nbsp; &nbsp; return content

&nbsp; &nbsp; def fail_open(self, error: Exception):
&nbsp; &nbsp; &nbsp; &nbsp; """
&nbsp; &nbsp; &nbsp; &nbsp; 失败开放策略:记录错误但不中断主流程

&nbsp; &nbsp; &nbsp; &nbsp; Args:
error: 异常对象
"""
&nbsp; &nbsp; &nbsp; &nbsp; print(f"注意: Langfuse hook&nbsp;error&nbsp;(non-blocking): {error}")
&nbsp; &nbsp; &nbsp; &nbsp; # 不抛出异常,保证主流程继续

关键设计点:

  1. 本地状态管理:记录上次处理位置,支持增量处理
  2. 文件锁机制:防止并发冲突
  3. 截断和哈希:处理大内容,生成唯一trace ID
  4. Fail-Open策略:监控失败不影响主流程

工程意义:一旦系统涉及多代理、长链路、外部搜索、知识库写入、源码修改,没有tracing几乎不可能稳定维护。

六、暴露的问题与改进方向

6.1 问题一:产品逻辑仍停留在Prompt层

README看起来强大,但真正落到程序层面的主要是:

  • 知识库管理器
  • Tracing钩子

而关键能力仍依赖Claude Code”照着Prompt自己做”:

  • 如何真正生成C/C++/Rust代码
  • 如何真正执行编译流水线
  • 如何将知识库组合映射到源码模板
  • 如何验证输出产物一致性

本质:更像带强约束的人工编排框架,而非deterministic pipeline。

6.2 问题二:知识库Schema漂移

// 问题示例:字段命名不统一
{
"techniques": [
&nbsp; &nbsp; {
"id":&nbsp;"T001",
"evasion_type":&nbsp;"api_obfuscation", &nbsp;// 有的用这个
"type":&nbsp;"api_obfuscation", &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;// 有的用这个(重复)
&nbsp; &nbsp; &nbsp; ...
&nbsp; &nbsp; }
&nbsp; ],
"loader_components": {
"storage_methods": [
&nbsp; &nbsp; &nbsp; {"id":&nbsp;"embedded", ...}, &nbsp; &nbsp;&nbsp;// 有的用描述性ID
&nbsp; &nbsp; &nbsp; {"id":&nbsp;"sto_009", ...}, &nbsp; &nbsp; &nbsp;// 有的用编号ID
&nbsp; &nbsp; &nbsp; ...
&nbsp; &nbsp; ]
&nbsp; }
}

长期风险:Schema漂移会导致Agent对字段理解依赖上下文猜测,而非明确约定。

6.3 问题三:产品化超前于工程收口

README写得像成熟产品,但实际稳定存在的只有:

  • .claude

    配置层

  • JSON知识库

  • Python管理脚本

  • Langfuse钩子

差距:离”真正稳固的插件产品”还有距离。

七、四大学习点

7.1 用三层抽象管理Prompt系统

┌─────────────────────────────────────────┐
│ &nbsp;Commands(用户入口) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;│
│ &nbsp;- 产品化命令接口 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;│
└─────────────────────────────────────────┘
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ↓
┌─────────────────────────────────────────┐
│ &nbsp;Agents(角色定义) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;│
│ &nbsp;- 人格边界 + 权限控制 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; │
└─────────────────────────────────────────┘
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ↓
┌─────────────────────────────────────────┐
│ &nbsp;Skills(执行工艺) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;│
│ &nbsp;- 分阶段SOP + 输出规范 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;│
└─────────────────────────────────────────┘

可迁移场景:代码审计、应急响应、威胁狩猎、自动化报告。

7.2 把易碎知识固化成结构化知识库

反模式:知识飘在对话上下文里,会话结束一切归零。

正模式:

// 知识库沉淀
{
"techniques": [...], &nbsp; &nbsp;&nbsp;// 发现的模式
"components": [...], &nbsp; &nbsp;&nbsp;// 提取的组件
"scenarios": [...] &nbsp; &nbsp; &nbsp;&nbsp;// 验证的组合
}

7.3 让高风险任务变成阶段化流程

C2 Evasion的Phase设计本质:将高不确定性智能体任务改造成接近流程引擎模式。

不确定性任务 → 分阶段拆解 → 每阶段明确输入输出 → 可审计追溯

7.4 考虑Tracing

常见错误路径:

在多数系统生命周期中,问题往往出现在这样的循环里:功能开发完成立即上线,初期运行一切正常;直到出现异常或故障时,团队才发现缺乏足够的监控与观测数据,无法快速定位根因。由于事后补充监控涉及代码改动、环境部署和数据回填,过程既耗时又复杂,最终导致问题解决滞后、排障成本居高不下。

正确路径(本项目):

在系统的设计阶段就预先集成 Tracing 能力,是构建可观测性体系的关键起点。将追踪逻辑嵌入架构设计中,而不是事后补充,可在后续开发和上线阶段自然沉淀全链路数据。当应用进入生产环境时,每一次调用、延迟与异常都能被完整记录和关联。这样,当出现性能瓶颈或故障时,团队无需重现现场,就能依靠 Tracing 信息快速回溯调用路径、定位根因,实现真正的“问题可追溯、治理可量化”。

总结:

Evasion SubAgents项目的真正价值不在于某个具体规避技巧,而在于它已经在尝试回答一个更大的问题:

当安全研究开始被Agent化之后,我们究竟应该如何组织Prompt、知识、工具和执行记录?


内部CTF课程上线,总课程30+小时,优惠折扣中!

帮会简介

《安全渗透感知》是FreeBuf知识大陆的重量级帮会,帮会致力于漏洞POC/EXP、红队攻防实战,是系统化从基础入门到实战漏洞挖掘的教程社区,包含团队自整的挖掘注意点和案例,还包含分享的渗透经验、SRC漏洞案例、代码审计、挖洞思路等高价值资源。

内容框架(持续新增中)

目前已有「670+」小伙伴加入了帮会

加入方式目前帮会成员670+人,永久会员优惠后只需69.9元。

随着人数的增加及资源的积累,之后永久会员将涨价至99元。

有意向的师傅们可以扫码加入我们,共同进步。

如何加入帮会?→ 安卓/苹果用户可扫码使用优惠券↓↓

→ PC端用户可复制此链接到浏览器↓↓

https://wiki.freebuf.com/societyDetail?society_id=184

已加入帮会的小伙伴

可以加帮主进帮会内部交流群

请备注:帮会


免责声明:

本文所载程序、技术方法仅面向合法合规的安全研究与教学场景,旨在提升网络安全防护能力,具有明确的技术研究属性。

任何单位或个人未经授权,将本文内容用于攻击、破坏等非法用途的,由此引发的全部法律责任、民事赔偿及连带责任,均由行为人独立承担,本站不承担任何连带责任。

本站内容均为技术交流与知识分享目的发布,若存在版权侵权或其他异议,请通过邮件联系处理,具体联系方式可点击页面上方的联系我

本文转载自:C4安全 Arthur Arthur《AI免杀:基于 Evasion‑SubAgents 的Skills设计与实现》

推荐两个远程控制开发专栏 网络安全文章

推荐两个远程控制开发专栏

文章总结: 本文推荐两个远程控制开发专栏:用AI带你从零开发大型商业远程控制软件和安全工程专栏导航从银狐源码学安全工程。内容为软文推广,提供学习资源链接,适合对
评论:0   参与:  0