"""«📜 Логи»: pick a container, then its tail, your own grep or a question to the agent (SPEC L-1…L-6, L-10…L-17).""" import asyncio import html import logging from datetime import timedelta from enum import StrEnum from aiogram import Bot, F, Router from aiogram.enums import ParseMode from aiogram.exceptions import TelegramAPIError from aiogram.filters.callback_data import CallbackData from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.types import BufferedInputFile, CallbackQuery, InlineKeyboardMarkup, Message from aiogram.utils.keyboard import InlineKeyboardBuilder from app.access.service import AccessService from app.agent.service import Agent, Outcome from app.bot.menu import Nav, NavAction, free_input from app.db.models import LogMode, Role, User, utcnow from app.logs.grep import GrepError, GrepErrorCode, GrepResult, parse_grep, run, to_command from app.logs.journal import LogJournal from app.runtime.docker import Container, DockerClient, DockerError from app.texts import LogsTexts, Texts log = logging.getLogger(__name__) TAILS = (50, 200) # lines (L-2) PERIODS = (5, 60, 24 * 60) # minutes (L-2) GREP_WINDOW = timedelta(hours=24) MAX_MESSAGE = 3500 # leave room under Telegram's 4096 for the header and HTML markup MAX_FILE = 5 * 1024 * 1024 # L-5 class LogsAct(StrEnum): LIST = "l" OPEN = "o" TAIL = "t" SINCE = "s" GREP = "g" AGENT = "a" class LogsCb(CallbackData, prefix="lg"): a: LogsAct id: str = "" # 12-char container id prefix n: int = 0 # lines for TAIL, minutes for SINCE class LogsFlow(StatesGroup): grep = State() agent = State() def period(t: LogsTexts, minutes: int) -> str: return t.hours.format(n=minutes // 60) if minutes >= 60 else t.minutes.format(n=minutes) def list_keyboard(texts: Texts, containers: list[Container]) -> InlineKeyboardMarkup: kb = InlineKeyboardBuilder() for c in containers: kb.button(text=f"{c.light.value} {c.name}", callback_data=LogsCb(a=LogsAct.OPEN, id=c.short_id)) kb.button(text=texts.menu.back, callback_data=Nav(action=NavAction.BACK)) kb.adjust(1) return kb.as_markup() def container_keyboard(texts: Texts, short_id: str) -> InlineKeyboardMarkup: t = texts.logs kb = InlineKeyboardBuilder() for n in TAILS: kb.button(text=t.tail.format(n=n), callback_data=LogsCb(a=LogsAct.TAIL, id=short_id, n=n)) for minutes in PERIODS: kb.button(text=period(t, minutes), callback_data=LogsCb(a=LogsAct.SINCE, id=short_id, n=minutes)) kb.button(text=t.grep_button, callback_data=LogsCb(a=LogsAct.GREP, id=short_id)) kb.button(text=t.agent_button, callback_data=LogsCb(a=LogsAct.AGENT, id=short_id)) kb.button(text=t.to_list, callback_data=LogsCb(a=LogsAct.LIST)) kb.adjust(2, 3, 2, 1) return kb.as_markup() def notes(t: LogsTexts, result: GrepResult) -> list[str]: out = [] if result.truncated: out.append(t.truncated.format(n=len(result.lines))) if result.partial: out.append(t.partial) if result.skipped: out.append(t.skipped.format(n=result.skipped)) return out def grep_error(t: LogsTexts, e: GrepError) -> str: template = { GrepErrorCode.EMPTY: t.err_empty, GrepErrorCode.QUOTES: t.err_quotes, GrepErrorCode.FLAG: t.err_flag, GrepErrorCode.CONTEXT: t.err_context, GrepErrorCode.EXTRA: t.err_extra, GrepErrorCode.TOO_LONG: t.err_too_long, GrepErrorCode.REGEX: t.err_regex, }[e.code] return template.format(arg=html.escape(e.arg[:100])) async def send_result( msg: Message, texts: Texts, head: str, body: str, footer: list[str], kb: InlineKeyboardMarkup, filename: str ) -> None: """Short results inline as
, long ones as a .log document (L-5); the container keyboard rides along."""
t = texts.logs
tail = "".join(f"\n{html.escape(n)}" for n in footer)
if not body:
await msg.answer(f"{head}\n{t.nothing}{tail}", parse_mode=ParseMode.HTML, reply_markup=kb)
return
escaped = html.escape(body)
if len(escaped) <= MAX_MESSAGE:
await msg.answer(f"{head}\n{escaped}{tail}", parse_mode=ParseMode.HTML, reply_markup=kb)
return
data = body.encode()[-MAX_FILE:]
await msg.answer_document(
BufferedInputFile(data, filename),
caption=f"{head}\n{t.sent_as_file}{tail}",
parse_mode=ParseMode.HTML,
reply_markup=kb,
)
async def deliver(
msg: Message, texts: Texts, head: str, count_only: bool, result: GrepResult, kb: InlineKeyboardMarkup, name: str
) -> None:
footer = notes(texts.logs, result)
if count_only: # -c: the number in the header is the answer
tail = "".join(f"\n{html.escape(n)}" for n in footer)
await msg.answer(head + tail, parse_mode=ParseMode.HTML, reply_markup=kb)
return
await send_result(msg, texts, head, "\n".join(result.lines), footer, kb, f"{name}-grep.log")
def who(user: User) -> str:
return f"{user.full_name}" + (f" @{user.username}" if user.username else "") + f" ({user.tg_id})"
def build_router(
texts: Texts,
docker: DockerClient | None,
journal: LogJournal,
agent: Agent | None = None,
access: AccessService | None = None,
) -> Router:
router = Router(name="logs")
t = texts.logs
def need_docker() -> DockerClient:
if docker is None:
raise DockerError("docker is not configured (DOCKER_URL)")
return docker
async def alert_admins(bot: Bot, member: User, prompt: str, stage: str, reason: str) -> None:
"""🛡 to the admins only (SPEC AL-2, L-12). The asker sees just a polite refusal."""
if access is None:
return
text = t.attack_alert.format(
who=html.escape(who(member)),
stage=stage,
reason=html.escape(reason),
prompt=html.escape(prompt[:1000]),
)
for admin in await access.list_members():
if admin.role is not Role.ADMIN:
continue
try:
await bot.send_message(admin.tg_id, text, parse_mode=ParseMode.HTML)
except TelegramAPIError as e: # one blocked admin must not hide the alert from the others
log.warning("attack alert to %s failed: %s", admin.tg_id, e)
@router.message(F.text == texts.menu.logs)
async def open_logs(message: Message, state: FSMContext) -> None:
await state.clear()
try:
containers = await need_docker().containers()
except DockerError as e:
log.warning("logs: %s", e)
await message.answer(t.unavailable)
return
await message.answer(t.title if containers else t.empty, reply_markup=list_keyboard(texts, containers))
@router.callback_query(LogsCb.filter())
async def on_action(query: CallbackQuery, callback_data: LogsCb, state: FSMContext, member: User) -> None:
if not isinstance(query.message, Message):
await query.answer()
return
msg = query.message
try:
d = need_docker()
if callback_data.a is LogsAct.LIST:
await state.clear()
containers = await d.containers()
await msg.edit_text(t.title if containers else t.empty, reply_markup=list_keyboard(texts, containers))
await query.answer()
return
c = await d.get(callback_data.id)
kb = container_keyboard(texts, c.short_id)
back = InlineKeyboardBuilder()
back.button(text=t.to_container, callback_data=LogsCb(a=LogsAct.OPEN, id=c.short_id))
match callback_data.a:
case LogsAct.OPEN:
await state.clear()
await msg.edit_text(t.container.format(name=c.name), reply_markup=kb)
case LogsAct.TAIL | LogsAct.SINCE:
await query.answer()
if callback_data.a is LogsAct.TAIL:
n = min(max(callback_data.n, 1), max(TAILS))
text = await d.logs(c.id, tail=n)
scope, q = t.tail.format(n=n), {"tail": n}
else:
minutes = min(max(callback_data.n, 1), max(PERIODS))
text = await d.logs(c.id, since=utcnow() - timedelta(minutes=minutes))
scope, q = t.since.format(period=period(t, minutes)), {"since_minutes": minutes}
lines = text.splitlines()
await journal.record(member.tg_id, c.name, LogMode.TAIL, q, len(lines))
head = t.head_tail.format(name=html.escape(c.name), scope=scope, lines=len(lines))
await send_result(msg, texts, head, "\n".join(lines), [], kb, f"{c.name}.log")
return
case LogsAct.GREP:
await state.set_state(LogsFlow.grep)
await state.update_data(logs_id=c.short_id)
await msg.answer(
t.ask_grep.format(name=html.escape(c.name)),
parse_mode=ParseMode.HTML,
reply_markup=back.as_markup(),
)
case LogsAct.AGENT:
if agent is None:
await query.answer(t.agent_off, show_alert=True)
return
left = max(agent.quota - await agent.used_today(member.tg_id, utcnow()), 0)
if left == 0:
await query.answer(t.agent_quota.format(quota=agent.quota), show_alert=True)
return
await state.set_state(LogsFlow.agent)
await state.update_data(logs_id=c.short_id)
await msg.answer(
t.agent_ask.format(name=html.escape(c.name), left=left, quota=agent.quota),
reply_markup=back.as_markup(),
)
except DockerError as e:
log.warning("logs: %s", e)
await query.answer(t.unavailable, show_alert=True)
return
await query.answer()
@router.message(LogsFlow.grep, F.text, free_input(texts))
async def got_grep(message: Message, state: FSMContext, member: User) -> None:
try:
query = parse_grep(message.text or "")
except GrepError as e:
await message.answer(grep_error(t, e), parse_mode=ParseMode.HTML)
return # stay in the state: fix and resend
data = await state.get_data()
try:
d = need_docker()
c = await d.get(data.get("logs_id", ""))
text = await d.logs(c.id, since=utcnow() - GREP_WINDOW)
except DockerError as e:
log.warning("logs: %s", e)
await message.answer(t.unavailable)
return
await state.clear()
result = await asyncio.to_thread(run, query, text)
await journal.record(member.tg_id, c.name, LogMode.GREP, query.to_dict(), result.matched)
head = t.head_grep.format(
command=html.escape(to_command(query)), name=html.escape(c.name), matched=result.matched
)
await deliver(message, texts, head, query.count, result, container_keyboard(texts, c.short_id), c.name)
@router.message(LogsFlow.agent, F.text, free_input(texts))
async def got_question(message: Message, state: FSMContext, member: User, bot: Bot) -> None:
if agent is None: # pragma: no cover - the button doesn't open this state without an agent
await state.clear()
return
prompt = message.text or ""
data = await state.get_data()
await state.clear()
try:
d = need_docker()
current = await d.get(data.get("logs_id", ""))
containers = await d.containers()
except DockerError as e:
log.warning("logs: %s", e)
await message.answer(t.unavailable)
return
await message.answer(t.agent_thinking)
answer = await agent.ask(member.tg_id, prompt, [c.name for c in containers], current.name)
kb = container_keyboard(texts, current.short_id)
match answer.outcome:
case Outcome.ATTACK:
await message.answer(t.agent_refused, reply_markup=kb)
await alert_admins(bot, member, prompt, answer.stage, answer.reason)
return
case Outcome.QUOTA:
await message.answer(t.agent_quota.format(quota=agent.quota), reply_markup=kb)
return
case Outcome.UNAVAILABLE:
await message.answer(t.agent_unavailable, reply_markup=kb)
return
case Outcome.NOT_UNDERSTOOD:
await message.answer(t.agent_not_understood, reply_markup=kb)
return
q = answer.query
assert q is not None
target = next(c for c in containers if c.name == q.container) # the validator allowed only these names
try:
text = await d.logs(target.id, since=utcnow() - timedelta(minutes=q.since_minutes))
except DockerError as e:
log.warning("logs: %s", e)
await message.answer(t.unavailable)
return
grep = q.to_grep()
result = await asyncio.to_thread(run, grep, text, q.limit)
await journal.record(member.tg_id, target.name, LogMode.AGENT, q.model_dump(), result.matched)
head = t.head_agent.format(
command=html.escape(to_command(grep)),
name=html.escape(target.name),
period=period(t, q.since_minutes),
matched=result.matched,
)
await deliver(
message, texts, head, q.count_only, result, container_keyboard(texts, target.short_id), target.name
)
return router