From 6d2417299072eb3824aa603d052674cf4df168d2 Mon Sep 17 00:00:00 2001 From: moliyu1101 <2644528429@qq.com> Date: Sat, 22 Aug 2026 19:14:11 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20Killsweep=20=E9=80=9A=E6=9D=80?= =?UTF-8?q?=E9=97=AD=E7=8E=AF=E2=80=94=E2=80=94=E5=91=BD=E4=B8=AD=E5=90=8E?= =?UTF-8?q?=E6=89=B9=E9=87=8F=E6=B4=BE=E5=8F=91=E5=B7=B2=E5=AE=9E=E8=AF=81?= =?UTF-8?q?=E5=8F=97=E5=BD=B1=E5=93=8D=E7=AB=99=E7=82=B9=E6=89=93=E6=B4=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 通杀确认后，把 affected_table 里 status=verified 的同类站点批量入队，走原生 Target→Worker→（Stage-0 预筛）→落库链路，让每个实证中招的站点逐步复现并产出独立 Finding；candidate 不入队、跳过 origin/重复/无效/敏感主机，受 KILLSWEEP_REPLAY_ENQUEUE_LIMIT 上限约束。新增 2 条单测。 --- app/orchestrator.py | 44 +++++++++++++++++-- tests/test_killsweep_closed_loop.py | 66 +++++++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 4 deletions(-) create mode 100644 tests/test_killsweep_closed_loop.py diff --git a/app/orchestrator.py b/app/orchestrator.py index fa251b2..3bf7e83 100644 --- a/app/orchestrator.py +++ b/app/orchestrator.py @@ -126,6 +126,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 +3324,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 +3361,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 +3413,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: diff --git a/tests/test_killsweep_closed_loop.py b/tests/test_killsweep_closed_loop.py new file mode 100644 index 0000000..556805e --- /dev/null +++ b/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() \ No newline at end of file From c89c6841b70991ada7ce40ac3fdb61c08841f63d Mon Sep 17 00:00:00 2001 From: moliyu1101 <2644528429@qq.com> Date: Sat, 22 Aug 2026 19:24:33 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20Escalation=20=E9=97=AD=E7=8E=AF?= =?UTF-8?q?=E6=94=B6=E5=8F=A3=E2=80=94=E2=80=94=E5=8D=87=E7=BA=A7=E7=A9=BA?= =?UTF-8?q?=E5=8F=91=E5=B0=84=E4=B8=8D=E8=90=BD=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 扩大危害深挖虽过显著性门槛仍可能返回空结果，空升级洞落库会污染报告。新增 escalation_guard.has_emission 守卫：升级结果无任一实证（poc/raw_request/raw_response/description 全空）即放弃落库，只记一条 escalate_skip 事件。新增 4 条单测。 --- app/orchestrator.py | 8 ++++++++ app/tools/escalation_guard.py | 17 +++++++++++++++++ tests/test_escalation_guard.py | 27 +++++++++++++++++++++++++++ 3 files changed, 52 insertions(+) create mode 100644 app/tools/escalation_guard.py create mode 100644 tests/test_escalation_guard.py diff --git a/app/orchestrator.py b/app/orchestrator.py index 3bf7e83..d68b9a7 100644 --- a/app/orchestrator.py +++ b/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 @@ -3598,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: diff --git a/app/tools/escalation_guard.py b/app/tools/escalation_guard.py new file mode 100644 index 0000000..9bf1005 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/tests/test_escalation_guard.py b/tests/test_escalation_guard.py new file mode 100644 index 0000000..2d278ee --- /dev/null +++ b/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": ""})) + self.assertTrue(has_emission({"raw_response": "admin", "poc": ""})) + + def test_all_emission_fields_considered(self): + # 守卫判定至少看这四个字段；只要一个非空即放行。 + self.assertEqual(set(_EMISSION_FIELDS), {"poc", "raw_request", "raw_response", "description"}) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file