feat: Killsweep 通杀闭环 + Escalation 升级空发射不落库

app/orchestrator.py

@@ -21,6 +21,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app import dedup
+from app.tools import escalation_guard
 from app.agents import collector
 from app.agents import intel as intel_lib
 from app.agents import playbook_router
@@ -126,6 +127,8 @@ def _escalation_is_significant(orig_severity: str, res: dict) -> bool:
 REVIEW_RETRY_BACKOFF = float(os.environ.get("REVIEW_RETRY_BACKOFF", "300"))
 TARGET_HEARTBEAT_INTERVAL = float(os.environ.get("TARGET_HEARTBEAT_INTERVAL", "30"))
 KILLSWEEP_DEDUP_SCAN_LIMIT = int(os.environ.get("KILLSWEEP_DEDUP_SCAN_LIMIT", "200"))
+# 通杀闭环：单次通杀最多把多少个「已实证受影响」的同类站点批量入队，防一次打爆队列。
+KILLSWEEP_REPLAY_ENQUEUE_LIMIT = int(os.environ.get("KILLSWEEP_REPLAY_ENQUEUE_LIMIT", "50"))
 # 同一目标因临时 LLM 错误回队的最大次数（内存级，不耗 retry_count）。
 # 超过则置 dead 收敛，避免模型持续抽风时目标无限回队空转。
 MAX_TRANSIENT_LLM_REQUEUE = int(os.environ.get("MAX_TRANSIENT_LLM_REQUEUE", "5"))
@@ -3322,12 +3325,17 @@ def _release_killsweep(fut: asyncio.Future) -> None:
             row.notes = res.get("notes", "")
             row.status = "done"
 
-            # 判定可通杀 + 实证验证成功 → 把那个同款站点入挖掘队列出货
+            # 通杀闭环：可通杀 → affected_table 里已实证的同类站点批量入队打洞（含 verified_url）。
             enq = ""
+            if res.get("is_killsweep") and affected_table:
+                added = await self._enqueue_killsweep_affected(
+                    session, task_id, affected_table, origin_host)
+                if added:
+                    enq = f"；已将 {added} 个已实证受影响站点批量入队打洞"
             if res.get("is_killsweep") and res.get("verified") and res.get("verified_url"):
-                added = await self._enqueue_killsweep_target(
-                    session, task_id, res["verified_url"], origin_host)
-                enq = "；已将验证成功的同款站点入队出货" if added else ""
+                if await self._enqueue_killsweep_target(
+                        session, task_id, res["verified_url"], origin_host):
+                    enq = f"{enq}；已将验证成功的同款站点入队出货"
             try:
                 await session.commit()
             except IntegrityError:
@@ -3354,6 +3362,10 @@ def _release_killsweep(fut: asyncio.Future) -> None:
                     "[产品指纹与已有记录冲突，已保留本条源漏洞分析]"
                 ).strip()
                 row.updated_at = _now()
+                # 通杀闭环（降级分支同样收口）：已实证同类受影响站点批量入队打洞。
+                if res.get("is_killsweep") and affected_table:
+                    await self._enqueue_killsweep_affected(
+                        session, task_id, affected_table, origin_host)
                 if res.get("is_killsweep") and res.get("verified") and res.get("verified_url"):
                     await self._enqueue_killsweep_target(
                         session, task_id, res["verified_url"], origin_host)
@@ -3402,6 +3414,31 @@ async def _enqueue_killsweep_target(self, session: AsyncSession, task_id: str,
             return False
         return True
 
+    async def _enqueue_killsweep_affected(self, session: AsyncSession, task_id: str,
+                                          affected_table: list, origin: str) -> int:
+        """通杀闭环：把 Hunter 已实证（status=verified）的同款受影响站点批量入队打洞。
+
+        复用原生 Target 队列 → Worker 挖掘 →（Stage-0 预筛）→ 落库链路，让每个实证
+        中招的同类站点逐步复现并产出独立 Finding。candidate 行不入队，避免空耗；
+        上限 KILLSWEEP_REPLAY_ENQUEUE_LIMIT 防止一次通杀打爆队列。返回实际入队数。
+        """
+        if not affected_table:
+            return 0
+        enqueued = 0
+        for item in affected_table:
+            if enqueued >= KILLSWEEP_REPLAY_ENQUEUE_LIMIT:
+                break
+            if not isinstance(item, dict):
+                continue
+            if str(item.get("status") or "") != "verified":
+                continue
+            url = str(item.get("url") or "").strip()
+            if not url:
+                continue
+            if await self._enqueue_killsweep_target(session, task_id, url, origin):
+                enqueued += 1
+        return enqueued
+
     def trigger_escalation(self, task_id: str, finding_id: str, orig_severity: str) -> bool:
         """AI accepted 后触发扩大危害深挖；finding 级 inflight 去重，单洞只打一次。"""
         if finding_id in self._escalation_inflight:
@@ -3562,6 +3599,13 @@ async def _persist_escalation_finding(self, task_id: str, target_id: str, origin
             "affected_scope": res.get("affected_scope", ""),
             "kill_chain": res.get("kill_chain", []),
         }
+        # Escalation 闭环收口：空发射（无任一实证）不落库，避免污染报告。
+        if not escalation_guard.has_emission(res):
+            async with SessionLocal() as session:
+                await self._log(session, "escalation", "escalate_skip",
+                                f"升级结果无实证，放弃落库（空升级洞）: {title[:80]}",
+                                finding_id=origin_finding_id)
+            return
         async with SessionLocal() as session:
             origin = await session.get(Finding, origin_finding_id)
             if origin is None:

app/tools/escalation_guard.py

@@ -0,0 +1,17 @@
+"""Escalation 闭环收口：升级结果空发射守卫。
+
+扩大危害深挖虽然过了显著性门槛，仍可能返回空结果（升级猎人空跑/未带回任何实证）。
+这类空升级洞若直接落库会污染报告。此守卫做保守判断：只要还带了任一实证
+（poc / raw_request / raw_response / description），都放行进评审；全部为空才判为
+空发射。全确定性、无副作用、可单测。
+"""
+from __future__ import annotations
+
+_EMISSION_FIELDS = ("poc", "raw_request", "raw_response", "description")
+
+
+def has_emission(res) -> bool:
+    """升级结果是否携带实证。返回 False 表示应放弃落库（空发射）。"""
+    if not isinstance(res, dict):
+        return False
+    return any(str(res.get(k) or "").strip() for k in _EMISSION_FIELDS)

tests/test_escalation_guard.py

@@ -0,0 +1,27 @@
+"""覆盖 Escalation 闭环收口：升级结果空发射守卫 has_emission。"""
+import unittest
+
+from app.tools.escalation_guard import _EMISSION_FIELDS, has_emission
+
+
+class TestEscalationGuard(unittest.TestCase):
+    def test_returns_false_when_not_a_dict(self):
+        self.assertFalse(has_emission(None))
+        self.assertFalse(has_emission("x"))
+
+    def test_empty_result_rejected(self):
+        self.assertFalse(has_emission({}))
+        self.assertFalse(has_emission({"poc": "   ", "description": ""}))
+
+    def test_any_emission_passes(self):
+        self.assertTrue(has_emission({"description": "接管管理后台"}))
+        self.assertTrue(has_emission({"poc": "<script>alert(1)</script>"}))
+        self.assertTrue(has_emission({"raw_response": "<html>admin</html>", "poc": ""}))
+
+    def test_all_emission_fields_considered(self):
+        # 守卫判定至少看这四个字段；只要一个非空即放行。
+        self.assertEqual(set(_EMISSION_FIELDS), {"poc", "raw_request", "raw_response", "description"})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_killsweep_closed_loop.py

@@ -0,0 +1,66 @@
+"""覆盖通杀闭环：_enqueue_killsweep_affected 批量派发已实证受影响站点入队打洞。
+
+规则：只入队 status=verified 的行；跳过 origin 自身、无效/敏感主机、重复 host；
+candidate 不入队；受 KILLSWEEP_REPLAY_ENQUEUE_LIMIT 上限约束。
+"""
+import unittest
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.db.models import Base, Target
+from app.orchestrator import TaskRunner
+
+ORIGIN = "https://https.test.school.edu.cn"
+_ORIGIN_HOST = "https.test.school.edu.cn"
+
+
+class TestKillsweepClosedLoop(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+        async with self.engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+        self.Session = async_sessionmaker(self.engine, expire_on_commit=False)
+
+    async def asyncTearDown(self):
+        await self.engine.dispose()
+
+    async def _hosts(self, task_id: str) -> set:
+        async with self.Session() as s:
+            rows = (await s.execute(
+                select(Target.host).where(Target.task_id == task_id)
+            )).scalars().all()
+        return set(rows)
+
+    @staticmethod
+    def _runner():
+        # 用一个不触发 __init__ 的空壳实例绑定目标方法，仅供单测调用。
+        runner = object.__new__(TaskRunner)
+        runner._enqueue_killsweep_target = TaskRunner._enqueue_killsweep_target.__get__(runner, TaskRunner)
+        return runner
+
+    async def test_verified_dispatched_candidate_and_origin_skipped(self):
+        table = [
+            {"url": "http://a.school.edu.cn", "status": "verified"},
+            {"url": "http://b.school.edu.cn", "status": "verified"},
+            {"url": "http://c.school.edu.cn", "status": "candidate"},  # 不入队
+            {"url": ORIGIN, "status": "verified"},                     # 源站自身，跳过
+            {"url": "http://a.school.edu.cn", "status": "verified"},   # 重复 host，跳过
+        ]
+        async with self.Session() as s:
+            count = await self._runner()._enqueue_killsweep_affected(
+                s, "task_t1", table, ORIGIN)
+            await s.commit()
+        self.assertEqual(count, 2)
+        self.assertEqual(await self._hosts("task_t1"), {"a.school.edu.cn", "b.school.edu.cn"})
+
+    async def test_empty_table_returns_zero(self):
+        async with self.Session() as s:
+            count = await self._runner()._enqueue_killsweep_affected(
+                s, "task_t2", [], ORIGIN)
+        self.assertEqual(count, 0)
+        self.assertEqual(await self._hosts("task_t2"), set())
+
+
+if __name__ == "__main__":
+    unittest.main()