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>
87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""Step 4 (SPEC L-13, L-14): the only thing the planner may produce, and its validation.
|
|
|
|
Even if the guard is fooled, the worst a planner reply can do is a slow grep over a managed container's logs.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
import regex
|
|
from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
|
|
|
|
from app.logs.grep import MAX_PATTERN, GrepQuery
|
|
|
|
MAX_MINUTES = 7 * 24 * 60
|
|
|
|
# Nested quantifiers and quantified alternations — the classic catastrophic-backtracking shapes: (a+)+, (a|aa)*
|
|
_NESTED = regex.compile(r"\((?:[^()\\]|\\.)*[+*}](?:[^()\\]|\\.)*\)\s*[+*{]")
|
|
_ALT_REPEAT = regex.compile(r"\((?:[^()\\]|\\.)*\|(?:[^()\\]|\\.)*\)\s*[+*{]")
|
|
|
|
|
|
def risky_backtracking(pattern: str) -> bool:
|
|
return bool(_NESTED.search(pattern) or _ALT_REPEAT.search(pattern))
|
|
|
|
|
|
class AgentLogQuery(BaseModel):
|
|
"""Validated with `context={"containers": [...]}` — the whitelist of names."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
container: str
|
|
since_minutes: int = Field(ge=1, le=MAX_MINUTES)
|
|
pattern: str = Field(min_length=1, max_length=MAX_PATTERN)
|
|
ignore_case: bool = True
|
|
invert: bool = False
|
|
count_only: bool = False
|
|
context: int = Field(default=0, ge=0, le=5)
|
|
limit: int = Field(default=100, ge=1, le=500)
|
|
|
|
@field_validator("container")
|
|
@classmethod
|
|
def _whitelisted(cls, value: str, info: ValidationInfo) -> str:
|
|
allowed = (info.context or {}).get("containers", [])
|
|
if value not in allowed:
|
|
raise ValueError("container is not in the list")
|
|
return value
|
|
|
|
@field_validator("pattern")
|
|
@classmethod
|
|
def _safe_regex(cls, value: str) -> str:
|
|
try:
|
|
regex.compile(value)
|
|
except regex.error as e:
|
|
raise ValueError(f"pattern does not compile: {e}") from e
|
|
if risky_backtracking(value):
|
|
raise ValueError("pattern may backtrack catastrophically")
|
|
return value
|
|
|
|
def to_grep(self) -> GrepQuery:
|
|
return GrepQuery(
|
|
pattern=self.pattern,
|
|
ignore_case=self.ignore_case,
|
|
invert=self.invert,
|
|
extended=True,
|
|
count=self.count_only,
|
|
context=self.context,
|
|
)
|
|
|
|
|
|
def json_schema() -> dict[str, Any]:
|
|
"""Strict structured-output schema for the planner: every field required, no extras.
|
|
Bounds are left to the pydantic model — strict mode of several providers rejects minimum/maxLength."""
|
|
types = {
|
|
"container": "string",
|
|
"since_minutes": "integer",
|
|
"pattern": "string",
|
|
"ignore_case": "boolean",
|
|
"invert": "boolean",
|
|
"count_only": "boolean",
|
|
"context": "integer",
|
|
"limit": "integer",
|
|
}
|
|
return {
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"required": list(types),
|
|
"properties": {name: {"type": kind} for name, kind in types.items()},
|
|
}
|