feat: 侦察期匿名泄露与 WAF 画像闭环 + 审核降过杀救援

app/agents/reviewer.py

@@ -272,6 +272,95 @@ def _maybe_deepen_ignored(finding: Finding, review: Review, src_type: str) -> bo
     return True
 
 
+def _evidence_fragmentary(finding: Finding) -> bool:
+    """证据残缺判定：缺一条可直接取证的高价值数据样本，且无响应包支撑。
+
+    用于匿名泄露救援——有 extracted_data_sample/raw_response 说明敏感数据已抓到，
+    证据并不残缺；仍被 ignored 属垃圾洞，不救。
+    """
+    if (finding.raw_response or "").strip():
+        return False
+    if (finding.evidence.extracted_data_sample or "").strip():
+        return False
+    return True
+
+
+def _has_some_evidence(finding: Finding) -> bool:
+    """证据至少一点：只要有任一可支撑的取证原料（响应/PoC/请求/样本/链路），就不算空洞。"""
+    if (finding.raw_response or "").strip():
+        return True
+    if (finding.raw_request or "").strip():
+        return True
+    if (finding.poc or "").strip():
+        return True
+    if (finding.evidence.extracted_data_sample or "").strip():
+        return True
+    if (finding.evidence.tool_output or "").strip():
+        return True
+    if finding.kill_chain and any((s.detail or "").strip() for s in finding.kill_chain):
+        return True
+    return False
+
+
+def _maybe_rescue_recon_anon_leak(finding: Finding, review: Review) -> bool:
+    """匿名泄露降过杀救援：侦察已实锤匿名端点泄露敏感数据，但中/低危被 ignored 且证据残缺
+    的发现是「证据残缺但高价值」线索，不当垃圾丢，改判 deepen 定向补证，避免漏报。"""
+    if review.verdict != ReviewVerdict.ignored or review.is_duplicate or not review.in_scope:
+        return False
+    if not getattr(finding.self_check, "recon_anon_leak", False):
+        return False
+    if finding.severity_claimed not in {Severity.medium, Severity.low}:
+        return False
+    if not _evidence_fragmentary(finding):
+        return False
+    review.verdict = ReviewVerdict.deepen
+    review.confidence = Confidence.uncertain
+    review.severity_final = None
+    review.score = min(max(review.score or 0, 2.5), 3.9)
+    review.ignore_reasons = []
+    review.deepen_directive = (
+        "该匿名端点已被侦察确认可公开访问且疑似返回敏感数据，但证据残缺。"
+        "请沿此端点用 curl 复现原始请求，脱敏截取响应体中对应的敏感字段（凭证/token/密钥/内部路径等），"
+        "给出请求包+响应关键片段，明确命中字段与敏感类型；确认并非页面正常展示内容、确有可取证敏感数据后提交。"
+        "若确认响应无敏感数据，finish no_vuln。"
+    )
+    review.reviewer_notes = (
+        (review.reviewer_notes or "").strip()
+        + "\n[系统改判] 该发现为侦察期匿名泄露线索、证据残缺但高价值，已由 ignored 改为 deepen 定向补证。"
+    ).strip()
+    return True
+
+
+def _maybe_rescue_recon_waf(finding: Finding, review: Review) -> bool:
+    """WAF 实锤降过杀救援：侦察已确认目标存在 WAF/风控拦截，这类发现常因请求被拦而证据残缺；
+    只要带 recon_waf_present 标记且证据至少一点，不直接 ignored，改判 deepen 并附绕过变形头。"""
+    if review.verdict != ReviewVerdict.ignored or review.is_duplicate or not review.in_scope:
+        return False
+    if not getattr(finding.self_check, "recon_waf_present", False):
+        return False
+    if not _has_some_evidence(finding):
+        return False
+    review.verdict = ReviewVerdict.deepen
+    review.confidence = Confidence.uncertain
+    review.severity_final = None
+    review.score = min(max(review.score or 0, 2.5), 3.9)
+    review.ignore_reasons = []
+    review.deepen_directive = (
+        "目标已确认存在 WAF/风控拦截，本次证据不足可能因请求被拦导致。"
+        "请沿同一注入点携带绕过变形头重试，先 baseline 后 variant 对比响应差异再实锤："
+        "- X-Forwarded-For: 127.0.0.1   /   X-Forwarded-For: 1.1.1.1, 127.0.0.1"
+        "- X-Real-IP: 127.0.0.1"
+        "- 随机正常 User-Agent / 空 User-Agent / Accept: text/html 等常见浏览器请求头"
+        "对比基线响应与变形后响应（状态码/拦截页/业务回显）差异，坐实注入是否被 WAF 放行；"
+        "拿到明确差异或业务回显后提交，若所有变形均被拦确证无真实注入则 finish no_vuln。"
+    )
+    review.reviewer_notes = (
+        (review.reviewer_notes or "").strip()
+        + "\n[系统改判] 该发现带侦察期 WAF 实锤、证据或残缺，已由 ignored 改为 deepen 并附绕过变形头定向深挖。"
+    ).strip()
+    return True
+
+
 def _is_thinking_tool_choice_error(err: LLMError) -> bool:
     """强制指定 submit_review 的 tool_choice 不被模型/网关接受时的兜底判定。
 
@@ -360,6 +449,10 @@ def review(self, finding: Finding) -> Review:
             self._emit("review_auto_ignore_weak_backdoor", title=finding.title)
         elif _maybe_accept_write_proof(finding, review):
             self._emit("review_auto_accept_write", title=finding.title, write_kind=classify_write_proof(finding))
+        elif _maybe_rescue_recon_anon_leak(finding, review):
+            self._emit("review_auto_rescue_recon_anon_leak", title=finding.title, directive=review.deepen_directive)
+        elif _maybe_rescue_recon_waf(finding, review):
+            self._emit("review_auto_rescue_recon_waf", title=finding.title, directive=review.deepen_directive)
         elif _maybe_deepen_ignored(finding, review, self.src_type):
             self._emit("review_auto_deepen", title=finding.title, directive=review.deepen_directive)
 

app/agents/worker.py

@@ -27,6 +27,7 @@
 from app.llm.client import LLMClient, LLMError, llm_error_event_fields
 from app.schemas import Finding, Verdict, WorkerResult
 from app.tools.executor import ToolExecutor
+from app.tools.anon_leak_scanner import scan_body as _leak_scan_body
 from app.tools.schemas import (
     JS_ANALYZER_TOOL_SCHEMAS,
     SESSION_TOOL_SCHEMAS,
@@ -45,6 +46,25 @@
     "下一条是目标/情报。只打当前目标；确认无攻击面才快速 finish。工具按信号开放，JS 线索再用 analyze_javascript。"
 )
 
+# ---- 侦察→挖洞闭环：匿名泄露嗅探 + WAF 画像前置 ----
+# 单个 worker 最多嗅探的匿名端点 / 上报的泄露线索条数（防一次侦察拖垮首轮）。
+_ANON_SCAN_LIMIT = int(os.environ.get("ANON_SCAN_LIMIT", "5"))
+_ANON_LEADS_EMIT = 3
+# 拦截样本上限；外科手术式只取最先命中的 WAF 指纹。
+_WAF_SAMPLE_LIMIT = 3
+# 是否补一次 live 基线探测来画像 WAF（即使侦察未给 recon_blocked，首包也能带变形策略）。
+_WAF_LIVE_PROBE = os.environ.get("WORKER_WAF_LIVE_PROBE", "1") == "1"
+_RECON_BLOCK_CODES = {"400", "401", "403", "406", "429", "501", "503"}
+# 命中这些类型的发现才把侦察锚点写进 self_check，供审核降过杀救援。
+_ANON_LEAK_VULN_TYPES = {
+    "unauthorized_access", "information_disclosure", "sensitive_information_disclosure",
+    "info_leak", "info_disclosure", "weak_password", "default_password",
+}
+_WAF_RELATED_VULN_TYPES = {
+    "sql_injection", "rce", "command_injection", "command_exec", "code_injection",
+    "ssti", "deserialization", "xss", "lfi", "rfi",
+}
+
 # LLM 调用失败时，worker 内软重试次数（清粘性后换端点/同端点再试），耗尽才整轮收尾回队。
 _WORKER_LLM_SOFT_RETRIES = int(os.environ.get("WORKER_LLM_SOFT_RETRIES", "3"))
 _WORKER_LLM_SOFT_RETRY_KINDS = {
@@ -160,6 +180,112 @@ def _intel_block(self) -> str:
         lines.append("提交漏洞时，请核实归属（域名/备案/证书CN/页脚版权/FOFA org/登录页品牌）后把最终归属写进 submit_finding 的 owner 字段。")
         return "\n".join(lines) + "\n\n" + self._user_auth_block() + self._creds_block() + self._intel_lib_block()
 
+    def scan_anon_leaks(self) -> list[dict]:
+        """对侦察期收集的匿名可访问端点做敏感数据泄露嗅探。
+
+        结果写 target_meta['anon_leak_leads']（供首轮注入与审核看板），命中即置
+        target_meta['recon_anon_leak']。全部走 executor，尽力而为，失败不阻断。
+        """
+        anon_open = list((self.target_meta or {}).get("anon_open") or [])
+        leads: list[dict] = []
+        for ep in anon_open[:_ANON_SCAN_LIMIT]:
+            url = str(ep.get("url") if isinstance(ep, dict) else ep or "").strip()
+            if not url:
+                continue
+            try:
+                resp = self.executor.http_request(url=url, method="GET")
+            except Exception:
+                resp = {"ok": False}
+            if not isinstance(resp, dict) or resp.get("ok") is not True:
+                continue
+            body = str(resp.get("body") or "")
+            hits = _leak_scan_body(body)
+            if hits:
+                leads.append({
+                    "url": url,
+                    "status": resp.get("status_code"),
+                    "hits": hits[:5],
+                    "snippet": body[:400],
+                })
+        self.target_meta["anon_leak_leads"] = leads[:_ANON_LEADS_EMIT]
+        if leads:
+            self.target_meta["recon_anon_leak"] = True
+        return leads
+
+    def profile_waf(self) -> dict:
+        """对拦截应答（含可选的 live 基线探测）做 WAF/风控指纹画像。
+
+        结果写 target_meta['waf_profile']，命中即置 target_meta['recon_waf_present']。
+        header_variants（X-Forwarded-For / X-Real-IP / UA 等）供首轮变形降被拦。
+        """
+        samples: list[dict] = list((self.target_meta or {}).get("recon_blocked") or [])
+        if _WAF_LIVE_PROBE:
+            try:
+                probe = self.executor.http_request(url=self.target, method="GET")
+            except Exception:
+                probe = None
+            if isinstance(probe, dict) and probe.get("ok") is True:
+                headers = (probe.get("response_headers") or {})
+                headers = headers if isinstance(headers, dict) else {}
+                samples.append({
+                    "status_code": probe.get("status_code"),
+                    "headers": headers,
+                    "body": str(probe.get("body") or ""),
+                })
+        profile: dict = {}
+        for s in samples[:_WAF_SAMPLE_LIMIT]:
+            sc = s.get("status_code")
+            if not isinstance(sc, int):
+                continue
+            info = self.executor.detect_waf_profiling(sc, s.get("headers") or {}, str(s.get("body") or ""))
+            if not isinstance(info, dict) or not info.get("ok") or not info.get("detected"):
+                continue
+            profile = {
+                "waf_type": info.get("waf_type") or "unknown",
+                "evidence": info.get("evidence") or "",
+                "blocked_likely": bool(info.get("blocked_likely")),
+                "strategy_priority": list(info.get("strategy_priority") or []),
+                "header_variants": list(info.get("header_variants") or []),
+            }
+            break
+        self.target_meta["waf_profile"] = profile
+        if profile.get("waf_type"):
+            self.target_meta["recon_waf_present"] = True
+        return profile
+
+    def _recon_leads_block(self) -> str:
+        """首轮注入块：侦察期匿名泄露线索 + WAF 画像（空则不输出任何内容）。"""
+        m = self.target_meta or {}
+        parts: list[str] = []
+        leaks = list(m.get("anon_leak_leads") or [])
+        if leaks:
+            lines = ["\n# 侦察期匿名泄露线索（首包携带，优先核实后再报）"]
+            for it in leaks:
+                lines.append(f"- {it.get('url')} 响应疑似泄露：{', '.join(it.get('hits') or [])}")
+            parts.append("\n".join(lines))
+        wp = m.get("waf_profile") or {}
+        if isinstance(wp, dict) and wp.get("waf_type"):
+            lines = ["\n# 目标 WAF/风控画像（首包即可针对性变形降低被拦）"]
+            lines.append(f"- 疑似 WAF：{wp['waf_type']}（{wp.get('evidence') or '本地指纹'}）")
+            hdrs = list(wp.get("header_variants") or [])
+            if hdrs:
+                lines.append("- 可尝试的绕过请求头（先 baseline 后 variant 对比差异再实锤）：")
+                for h in hdrs[:4]:
+                    lines.append(f"  - {', '.join(f'{k}={v}' for k, v in h.items())}")
+            parts.append("\n".join(lines))
+        return ("\n".join(parts) + "\n") if parts else ""
+
+    def _recon_probe(self) -> None:
+        """首次挖掘轮前置：匿名泄露嗅探 + WAF 画像。均为尽力而为，任何失败不阻断进场。"""
+        try:
+            self.scan_anon_leaks()
+        except Exception:
+            pass
+        try:
+            self.profile_waf()
+        except Exception:
+            pass
+
     def _user_auth_block(self) -> str:
         """用户在凭据区提供的 Cookie/账密（系统已尝试后的回执）。"""
         ctx = (self.target_meta or {}).get("user_auth") or (self.target_meta or {}).get("auth_context")
@@ -1279,6 +1405,17 @@ def _submit_finding(self, args: dict) -> dict:
                 ),
             }
 
+        # 侦察锚点：把侦察期已实锤的信号写进 self_check，供审核降过杀救援锚定。
+        # 只在命中对应漏洞类型时打标，避免无关发现被错误救援。
+        if (self.target_meta or {}).get("recon_anon_leak") and (
+            finding.vuln_type or ""
+        ).strip().lower() in _ANON_LEAK_VULN_TYPES:
+            finding.self_check.recon_anon_leak = True
+        if (self.target_meta or {}).get("recon_waf_present") and (
+            finding.vuln_type or ""
+        ).strip().lower() in _WAF_RELATED_VULN_TYPES:
+            finding.self_check.recon_waf_present = True
+
         duplicate, matches = self._dup_matches(finding.model_dump(mode="json"))
         if duplicate:
             self._emit("finding_duplicate", title=finding.title, matches=len(matches))

app/schemas.py

@@ -33,6 +33,8 @@ class SelfCheck(BaseModel):
     scanner_only_no_poc: bool = Field(False, description="是否仅扫描器出结果但无法给出利用方法（会被忽略）")
     is_public_interface: bool = Field(False, description="该接口是否本就是面向公众的公开接口（若是，访问它通常不构成漏洞）")
     info_leak_hits_strict_list: bool = Field(False, description="若属信息泄露类：泄露数据是否命中当前 SRC 模式的高价值敏感数据口径")
+    recon_anon_leak: bool = Field(False, description="发现是否源自侦察期匿名可访问端点泄露的敏感数据（审核降过杀救援锚点）")
+    recon_waf_present: bool = Field(False, description="侦察期已确认目标存在 WAF/风控拦截（审核降过杀救援锚点，这类发现常因被拦而证据残缺）")
 
 
 class Evidence(BaseModel):

app/tools/anon_leak_scanner.py

@@ -0,0 +1,42 @@
+"""侦察期匿名可访问端点敏感数据泄露嗅探（纯本地、无副作用、可单测）。
+
+对某个 HTTP 响应正文做保守的关键词/正则命中，识别“疑似泄露的敏感数据”，
+供 worker 在首轮就把线索带给 LLM 核实，并在审核侧作为降过杀的救援锚点。
+原则：只标记、不实锤；命中仅代表“值得核实”，最终判洞仍交由 worker 取证与审核把关。
+"""
+from __future__ import annotations
+
+import re
+
+# (标签, 正则)。各正则已内联 (?i)。保守起见按“长得像敏感数据”匹配。
+_LEAK_PATTERNS: list[tuple[str, str]] = [
+    ("令牌/Bearer", r"(?i)(authorization|bearer|x-api-key|api[_-]?key|token)['\"]?\s*[:=]\s*['\"]?(?:bearer\s+)?[a-z0-9._\-]{12,}['\"]?"),
+    ("密钥/口令", r"(?i)(password|passwd|pwd|secret)['\"]?\s*[:=]\s*['\"][^\s'\"<>]{6,}['\"]"),
+    ("私钥", r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
+    ("AWS 凭证", r"(?i)(AKIA[0-9A-Z]{16}|aws_access_key_id|aws_secret_access_key)"),
+    ("数据库连接串", r"(?i)(jdbc:|mysql://|postgres://|mongodb(\+srv)?://|redis://)[^\s'\"<>]+"),
+    ("对象存储端点", r"(?i)[a-z0-9.\-]+\.s3[.\-][a-z0-9\-]*\.amazonaws\.com"),
+    ("Slack/GitHub 令牌", r"(?i)(xox[baprs]-[0-9a-z\-]{10,}|gh[pousr]_[0-9a-z]{20,})"),
+    ("内存/SSN/身份证", r"(?i)(id_card|idcard|identity_card|ssn)['\"]?\s*[:=]\s*['\"][0-9]{6,}['\"]"),
+]
+
+_COMPILED: list[tuple[str, re.Pattern]] = [
+    (label, re.compile(pattern)) for label, pattern in _LEAK_PATTERNS
+]
+
+
+def scan_body(body: str) -> list[str]:
+    """在响应正文中命中疑似敏感数据，返回去重后的命中标签列表（可能为空）。"""
+    if not body:
+        return []
+    text = body if isinstance(body, str) else str(body)
+    hits = [label for label, rx in _COMPILED if rx.search(text)]
+    seen: list[str] = []
+    for label in hits:
+        if label not in seen:
+            seen.append(label)
+    return seen
+
+
+def has_sensitive_leak(body: str) -> bool:
+    return bool(scan_body(body))

app/tools/executor.py

@@ -783,3 +783,34 @@ def suggest_waf_bypass(
             )
         except Exception as e:
             return {"ok": False, "error": f"WAF 建议生成异常: {type(e).__name__}: {e}"}
+
+    def detect_waf_profiling(
+        self,
+        status_code: int,
+        response_headers: Optional[dict[str, Any]] = None,
+        response_body: str = "",
+    ) -> dict[str, Any]:
+        """对单个被拦截应答做本地 WAF/风控指纹画像（纯本地、不发网络）。
+
+        供 worker 首轮前置画像目标 WAF，产出的 header_variants（X-Forwarded-For /
+        X-Real-IP / UA 等）可直接用于首包针对性变形降低被拦概率。
+        """
+        try:
+            info = _suggest_waf_bypass(
+                payload="''",
+                status_code=status_code,
+                response_headers=response_headers,
+                response_body=response_body,
+                context="generic",
+            )
+            return {
+                "ok": True,
+                "detected": bool(info.get("detected")),
+                "waf_type": info.get("waf_type"),
+                "evidence": info.get("evidence", ""),
+                "blocked_likely": bool(info.get("blocked_likely")),
+                "strategy_priority": list(info.get("strategy_priority") or []),
+                "header_variants": list(info.get("header_variants") or []),
+            }
+        except Exception as e:
+            return {"ok": False, "error": f"WAF 画像生成异常: {type(e).__name__}: {e}"}