ElebobaandClaude Opus 5.5
65acf3abf1
CI / ruff · mypy · pytest (pull_request) Successful in 39m13s
- prefilter: правила из config/agent_rules.yaml (длина, команды, shell,
секреты и системные файлы, prompt injection) — без LLM
- guard: отдельный вызов, JSON {verdict, reason}; невалидный ответ = attack,
недоступный LLM — «попробуйте позже», а не атака
- planner: только AgentLogQuery (контейнер из белого списка, ≤ 7 д, шаблон
≤ 200 без вложенных квантификаторов, лимит ≤ 500); логи в LLM не идут
- бот: «🤖 Спросить агента» → показывает построенный grep и строки;
атака — вежливый отказ и 🛡 администраторам; квота 30 в сутки
- журнал agent_requests + log_queries (mode=agent)
- при LLM_PROVIDER=mock агент выключен
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
"""A fake LLM that answers per stage: the guard and the planner get the same user text, so MockProvider's
|
|
`contains` rules can't tell them apart."""
|
|
|
|
from typing import Any
|
|
|
|
from app.llm.provider import ChatMessage, JsonSchema, LLMError, LLMResult
|
|
|
|
|
|
class ScriptedLLM:
|
|
def __init__(self, guard: dict[str, Any] | str | None = None, plan: dict[str, Any] | str | None = None) -> None:
|
|
self.guard: dict[str, Any] | str | None = guard if guard is not None else {"verdict": "safe", "reason": "ok"}
|
|
self.plan = plan
|
|
self.down = False
|
|
self.calls: list[tuple[str, list[ChatMessage]]] = []
|
|
|
|
async def complete(self, messages: list[ChatMessage], schema: JsonSchema | None = None) -> LLMResult:
|
|
stage = schema.name if schema else "text"
|
|
self.calls.append((stage, messages))
|
|
if self.down:
|
|
raise LLMError("connection refused")
|
|
reply = self.guard if stage == "guard_verdict" else self.plan
|
|
if isinstance(reply, dict):
|
|
return LLMResult(text=str(reply), data=reply, prompt_tokens=10, completion_tokens=5)
|
|
return LLMResult(text=reply or "", data=None, prompt_tokens=10, completion_tokens=5)
|
|
|
|
async def aclose(self) -> None:
|
|
pass
|
|
|
|
def stages(self) -> list[str]:
|
|
return [s for s, _ in self.calls]
|
|
|
|
|
|
def plan(**overrides: Any) -> dict[str, Any]:
|
|
base: dict[str, Any] = {
|
|
"container": "crash",
|
|
"since_minutes": 60,
|
|
"pattern": "connection|refused",
|
|
"ignore_case": True,
|
|
"invert": False,
|
|
"count_only": False,
|
|
"context": 0,
|
|
"limit": 100,
|
|
}
|
|
return {**base, **overrides}
|