"""The AI's triage read: 「是不是真的有问题」. The security desk has to answer that before anything else, and doing it for 149 reports of somebody else's English is the mechanical reason 118 never got an answer. These cover the rules that make the answer safe to show — not the model's taste, which is not testable here, but what happens to a reply that is wrong, malformed, or absent. """ from __future__ import annotations from typing import Any import pytest from app.domains.bounty import ai as bounty_ai pytestmark = pytest.mark.asyncio REAL_ENOUGH = ( "Title: Improper access control on /dashboard/api/statements/plan/binding\n" "Description: a read-only shared session can create and delete plan " "bindings, which should be GET-only.\n" "Steps: share the session read-only, then press Bind on a statement.\n" "Impact: a view-only user can change bindings." ) class _Stub: """Stands in for the Bailian client. The real call is exercised the first time somebody presses 「让 AI 初筛」 after deploy; what has to be tested here is how a REPLY is handled, which is the part that can be wrong without anybody noticing.""" model = "qwen-stub" def __init__(self, payload: Any) -> None: self._payload = payload self.calls = 0 async def chat_json(self, **_kwargs: Any) -> Any: self.calls += 1 return self._payload def _use(monkeypatch: pytest.MonkeyPatch, payload: Any) -> _Stub: stub = _Stub(payload) async def _build(_db: Any) -> _Stub: return stub monkeypatch.setattr(bounty_ai, "_build_client", _build) return stub async def _assess(db, **over: Any) -> dict[str, Any] | None: return await bounty_ai.assess_report( db, title=over.pop("title", "a finding"), raw_text=over.pop("raw_text", REAL_ENOUGH), severity_keys=over.pop("severity_keys", ["critical", "high", "medium"]), vuln_types=over.pop("vuln_types", ["idor", "xss"]), ) async def test_a_good_reply_comes_through_whole(db_session, monkeypatch): _use(monkeypatch, { "verdict": "likely_valid", "verdict_reason": "只读会话能改绑定,越过了它自己声明的权限边界。", "confidence": "high", "exploitability": "需要拿到一个只读分享链接,不需要额外权限。", "impact": "只读用户可以创建和删除执行计划绑定,破坏完整性。", "what_to_check": ["用只读分享码调用该端点,看是否 200", "确认 v8.5.7 仍未加校验"], "questions": [], "suggested_severity": "medium", "vuln_type": "idor", "affected_urls": ["/dashboard/api/statements/plan/binding"], "reproduction": "share read-only, press Bind", }) out = await _assess(db_session) assert out is not None assert out["verdict"] == "likely_valid" assert out["confidence"] == "high" assert out["suggested_severity"] == "medium" assert out["vuln_type"] == "idor" assert len(out["what_to_check"]) == 2 assert out["model"] == "qwen-stub" async def test_an_unknown_verdict_never_reads_as_a_clean_bill_of_health( db_session, monkeypatch, ): """The safety rule. A model that answers "fine" or "n/a" or invents a fifth verdict must not land on the page as anything a reader could take for "nothing to see here" — the honest fallback is "we could not tell".""" for bogus in ("fine", "valid", "", "NOT_A_PROBLEM", None): _use(monkeypatch, {"verdict": bogus, "verdict_reason": "x" * 20}) out = await _assess(db_session) assert out is not None assert out["verdict"] == "needs_info", bogus async def test_a_severity_outside_the_ladder_is_dropped( db_session, monkeypatch, ): """The tier carries money. A level the ladder does not contain must not reach the report, even as a suggestion.""" _use(monkeypatch, { "verdict": "likely_valid", "suggested_severity": "catastrophic", "vuln_type": "not-a-type", }) out = await _assess(db_session) assert out["suggested_severity"] is None assert out["vuln_type"] is None async def test_a_title_alone_is_never_assessed(db_session, monkeypatch): """0 of 149 production reports have a body. Guessing a verdict from a title is exactly how a critical finding gets called informational — and it would look identical to a real reading on the page.""" stub = _use(monkeypatch, {"verdict": "likely_valid"}) assert await _assess(db_session, raw_text="SQL injection") is None assert stub.calls == 0, "and it must not even call the model" async def test_a_broken_reply_yields_nothing_rather_than_something( db_session, monkeypatch, ): """A missing assessment is honest. A fabricated one on a security finding is worse than none, and this runs in bulk over 149 reports, so one bad reply must not stop the rest.""" for payload in (None, [], "sorry I cannot help", 42): _use(monkeypatch, payload) assert await _assess(db_session) is None, payload async def test_the_model_being_down_is_not_an_error(db_session, monkeypatch): from app.core.exceptions import AppException async def _boom(_db: Any) -> Any: raise AppException( code="BAILIAN_LLM_NO_KEY", message="no key", status_code=503, ) monkeypatch.setattr(bounty_ai, "_build_client", _boom) assert await _assess(db_session) is None async def test_the_lists_are_capped(db_session, monkeypatch): """`what_to_check` is a short checklist somebody actually works through. Twenty items is a wall, and a wall does not get read.""" _use(monkeypatch, { "verdict": "needs_info", "what_to_check": [f"step {n}" for n in range(20)], "questions": [f"q {n}" for n in range(20)], "affected_urls": [f"https://x/{n}" for n in range(30)], }) out = await _assess(db_session) assert len(out["what_to_check"]) == 4 assert len(out["questions"]) == 4 assert len(out["affected_urls"]) == 10 async def test_the_reading_is_written_in_the_readers_language( db_session, monkeypatch, ): """Hardcoding Chinese put Chinese prose under English headings for anyone whose console renders in English — half-translated is worse than either language on its own.""" seen: dict[str, str] = {} class _Capture(_Stub): async def chat_json(self, **kwargs: Any) -> Any: seen["user"] = kwargs.get("user", "") return await super().chat_json(**kwargs) stub = _Capture({"verdict": "needs_info"}) async def _build(_db: Any) -> Any: return stub monkeypatch.setattr(bounty_ai, "_build_client", _build) for lang, needle in ( ("en-US", "Answer in English"), ("ja", "日本語"), ("zh-CN", "简体中文"), (None, "简体中文"), # the security desk's default ("de", "简体中文"), # unknown locale falls back, never blank ): out = await bounty_ai.assess_report( db_session, title="t", raw_text=REAL_ENOUGH, severity_keys=["high"], vuln_types=["idor"], lang=lang, ) assert needle in seen["user"], (lang, seen["user"][:80]) assert out["lang"] == (lang or "zh").split("-")[0].lower()