1
class PlannerContext(BaseModel):
2
    """规划器上下文对象，保存历史规划信息"""
3
    planning_history: List[Dict[str, Any]] = []        # 所有规划尝试的完整记录
4
    rejected_strategies: Dict[str, str] = {}            # 被拒绝的策略及原因
5
    long_term_objectives: List[str] = []                # 长期战略目标
6
    latest_reflection_report: Optional[Dict[str, Any]]  # 最新反思报告
7
    previous_planning_session: Optional[Dict[str, Any]] # 上一次规划会话
8

9
class ReflectorContext(BaseModel):
10
    """反思器上下文对象，保存历史反思信息"""
11
    reflection_log: List[Dict[str, Any]] = []           # 反思日志
12
    validated_patterns: List[Dict[str, Any]] = []       # 已验证的攻击模式
13
    persistent_insights: List[Dict[str, Any]] = []      # 持久化洞察

1
class ContextCompressor:
2
    """
3
    基于三种策略进行智能压缩：
4
    1. 消息数量阈值（默认 50 条）
5
    2. 执行轮次阈值（每 10 轮）
6
    3. 估算 token 超限（默认 100000）
7
    """

1
def should_compress(self, messages, execution_count) -> tuple[bool, str]:
2
    # 策略 1: 消息数量 > 50
3
    if len(messages) > self.message_threshold:
4
        return True, f"消息数量过多"
5
    # 策略 2: 每 10 轮定期压缩
6
    if execution_count > 0 and execution_count % self.compress_interval == 0:
7
        return True, f"定期压缩"
8
    # 策略 3: 估算 token > 100000
9
    estimated_tokens = total_chars // 4
10
    if estimated_tokens > self.token_threshold:
11
        return True, f"估算 token 超限"

1
STE 结构:
2
├── 战略原则（Strategic Principle）：最高层次的攻击原则
3
├── 战术手册（Tactical Manual）：有序的、抽象的步骤列表
4
└── 适用场景（Applicable Scenarios）：标签列表，定义复用场景

1
class AntiAddiction:
2
    def __init__(self,
3
        similarity_threshold: float = 0.85,  # 相似度阈值
4
        max_similar_count: int = 18,          # 最大连续相似次数
5
        command_timeout: int = 300,            # 命令超时 (秒)
6
        max_retries: int = 3                   # 最大重试次数
7
    ):

1
def _is_similar_to_last_command(self, current_cmd: str) -> bool:
2
    # 1. 命令太短 (< 10 字符) 不检测
3
    if len(current_cmd) < self._min_cmd_length:
4
        return False
5
    # 2. 用 SequenceMatcher 计算与上一条命令的相似度
6
    similarity = SequenceMatcher(None, current_cmd, last_cmd).ratio()
7
    # 3. 相似度 >= 0.85 判定为相似命令
8
    return similarity >= self.similarity_threshold

1
class CommandHistory(BaseModel):
2
    commands: List[str] = []
3
    max_length: int = 18  # 18 条命令的滑动窗口
4

5
    def add(self, command: str) -> None:
6
        self.commands.append(command)
7
        if len(self.commands) > self.max_length:
8
            self.commands.pop(0)  # FIFO 弹出最早的命令

1
⚠️ 检测到可能陷入循环（已连续执行 18 次相似命令）
2

3
请先思考以下问题来重新制定计划：
4
1. 我的核心假设是什么？
5
2. 过去 18 次的尝试，是否证明了这个假设是错误的？
6
3. 除了当前的方法，还有哪些其他的可能性？
7
4. 是否有更高效的方式（如批量处理、自动化脚本）？

1
def create_retry_decorator(self, func_name: str = "") -> Callable:
2
    return retry(
3
        retry=retry_if_exception_type(Exception),
4
        wait=wait_exponential_jitter(
5
            initial=0.3,   # 首次重试等待 0.3 秒
6
            max=10,         # 最大等待 10 秒
7
            jitter=0.5      # ±0.5 秒随机抖动
8
        ),
9
        stop=stop_after_attempt(self.max_retries),  # 最多重试 3 次
10
        reraise=True,
11
    )

1
@staticmethod
2
def is_retryable_error(exception) -> bool:
3
    # ConnectionError → 可重试
4
    # TimeoutError → 可重试
5
    # HTTP 408, 404, 429, 500-526 → 可重试

1
async def execute_with_timeout(self, coro, timeout=None, func_name=""):
2
    timeout = timeout or self.command_timeout  # 默认 300 秒
3
    return await asyncio.wait_for(coro, timeout=timeout)

1
if tool_name == "execute_python_code":
2
    @retry_decorator
3
    async def _execute():
4
        outputs = self.python_executor.execute_code(
5
            session_name=arguments.get("session_name", "default"),
6
            code=arguments.get("code", ""),
7
            timeout=arguments.get("timeout", 120)
8
        )
9
        return "\n\n".join(
10
            f"Output: {output.get('text', '')}"
11
            for output in outputs
12
        )

1
elif tool_name == "execute_shell_command":
2
    @retry_decorator
3
    async def _execute():
4
        session_id = self.terminal.new_session()           # 创建新终端会话
5
        self.terminal.send_keys(session_id=session_id,     # 发送命令
6
            keys=arguments.get("command", ""), enter=True)
7
        await asyncio.sleep(2)                              # 等待执行
8
        output = self.terminal.get_output(session_id)       # 获取输出
9
        return output

1
self.proxy = Proxy(
2
    url=config.proxy_caido_url,
3
    token=config.proxy_caido_token
4
) if config.proxy_caido_token else None

1
- CyberspaceMapper    # 空间测绘（FOFA/Quake）
2
- FingerprintScanner  # 指纹识别
3
- DirectoryScanner    # 目录扫描
4
- JSAnalyzer          # JS 分析
5
- AuthBypassFuzzer    # 鉴权绕过 Fuzz
6
- ReconWorkflow       # 侦察工作流编排