139 lines
5.8 KiB
Python
139 lines
5.8 KiB
Python
import asyncio
|
|
import inspect
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from datetime import timedelta
|
|
from typing import Callable
|
|
|
|
from ..agent_runs import use_origin
|
|
from ..chat import callback_data
|
|
from ..chat.callback_data import ACTION_CANCEL, ACTION_CONFIRM, ACTION_DEEP, ACTION_DOWNTIME, ACTION_MUTE
|
|
from ..globs import literal_glob
|
|
from ..mutes import MAX_MUTE_MINUTES
|
|
from ..ratelimit import FailureLimiter
|
|
from .access import is_allowed
|
|
from .constants import (OUTCOME_DENIED, OUTCOME_ERROR, OUTCOME_OK, OUTCOME_RATE_LIMITED, OUTCOME_UNKNOWN,
|
|
RATE_LIMIT_MAX, RATE_LIMIT_WINDOW)
|
|
from .ai import agent_reply
|
|
from .context import AuditEvent, CommandDeps
|
|
from .downtimecmd import start_downtime
|
|
from .models import Reply
|
|
from .mutecmd import create_mute
|
|
from .render import plain
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
ANSWER_DENIED = "Nicht erlaubt."
|
|
ANSWER_UNKNOWN = "Unbekannte Aktion."
|
|
ANSWER_GONE = "Alarm nicht mehr bekannt."
|
|
ANSWER_BUSY = "Zu viele Anfragen. Bitte kurz warten."
|
|
ANSWER_ERROR = "Das hat gerade nicht geklappt."
|
|
ANSWER_DONE = "Erledigt."
|
|
BUTTON_REASON = "per Button"
|
|
COMMAND_DEEP = "deep"
|
|
LIMITED_ACTIONS = frozenset({ACTION_DEEP})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IncomingCallback:
|
|
backend: str
|
|
sender_id: str
|
|
target: str
|
|
message_id: str
|
|
data: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CallbackResult:
|
|
answer: str
|
|
reply: Reply | None = None
|
|
clear_buttons: bool = False
|
|
|
|
|
|
class CallbackHandler:
|
|
def __init__(self, deps: CommandDeps, allowlist: Callable[[str], str | None],
|
|
audit: Callable[[AuditEvent], object], limiter: FailureLimiter | None = None):
|
|
self._deps = deps
|
|
self._allowlist = allowlist
|
|
self._audit = audit
|
|
self._limiter = limiter or FailureLimiter(RATE_LIMIT_MAX, RATE_LIMIT_WINDOW)
|
|
|
|
async def _record(self, event: IncomingCallback, action: str, outcome: str) -> None:
|
|
try:
|
|
result = self._audit(AuditEvent(event.backend, event.sender_id, f"button:{action}",
|
|
event.data, outcome))
|
|
if inspect.isawaitable(result):
|
|
await result
|
|
except Exception:
|
|
log.exception("callback audit failed")
|
|
|
|
async def handle(self, event: IncomingCallback) -> CallbackResult:
|
|
decoded = callback_data.decode(event.data)
|
|
action = decoded[0] if decoded else "?"
|
|
if not is_allowed(event.backend, event.sender_id, self._allowlist(event.backend)):
|
|
await self._record(event, action, OUTCOME_DENIED)
|
|
return CallbackResult(ANSWER_DENIED)
|
|
if decoded is None:
|
|
await self._record(event, action, OUTCOME_UNKNOWN)
|
|
return CallbackResult(ANSWER_UNKNOWN)
|
|
if action in LIMITED_ACTIONS:
|
|
key = f"{event.backend}:{event.sender_id}"
|
|
if self._limiter.blocked(key):
|
|
await self._record(event, action, OUTCOME_RATE_LIMITED)
|
|
return CallbackResult(ANSWER_BUSY)
|
|
self._limiter.record(key)
|
|
try:
|
|
result = await self._dispatch(event, action, decoded[1])
|
|
except Exception:
|
|
log.exception("callback %s failed", action)
|
|
await self._record(event, action, OUTCOME_ERROR)
|
|
return CallbackResult(ANSWER_ERROR)
|
|
await self._record(event, action, OUTCOME_OK if result is not None else OUTCOME_UNKNOWN)
|
|
return result or CallbackResult(ANSWER_UNKNOWN)
|
|
|
|
async def _alert(self, parts: tuple[str, ...]) -> tuple[str, str] | None:
|
|
if not parts or not parts[0].isdecimal() or self._deps.lookup is None:
|
|
return None
|
|
return await asyncio.to_thread(self._deps.lookup.service_for_alert, int(parts[0]))
|
|
|
|
@staticmethod
|
|
def _minutes(parts: tuple[str, ...]) -> timedelta | None:
|
|
if len(parts) < 2 or not parts[1].isdecimal() or not 0 < int(parts[1]) <= MAX_MUTE_MINUTES:
|
|
return None
|
|
return timedelta(minutes=int(parts[1]))
|
|
|
|
async def _dispatch(self, event: IncomingCallback, action: str, parts: tuple[str, ...]) -> CallbackResult | None:
|
|
if action in (ACTION_CONFIRM, ACTION_CANCEL):
|
|
return await self._resolve(event, parts, action == ACTION_CONFIRM)
|
|
if action not in (ACTION_MUTE, ACTION_DEEP, ACTION_DOWNTIME):
|
|
return None
|
|
service = await self._alert(parts)
|
|
if service is None:
|
|
return CallbackResult(ANSWER_GONE)
|
|
host, name = service
|
|
if action == ACTION_DEEP:
|
|
with use_origin(event.backend, event.sender_id, COMMAND_DEEP):
|
|
result = await self._deps.agent.explain(host, name)
|
|
return CallbackResult(ANSWER_DONE, agent_reply(result))
|
|
duration = self._minutes(parts)
|
|
if duration is None:
|
|
return None
|
|
if action == ACTION_DOWNTIME:
|
|
if self._deps.downtime is None:
|
|
return CallbackResult(ANSWER_ERROR)
|
|
reply = await start_downtime(self._deps.downtime, event.backend, event.sender_id, host, name,
|
|
duration, BUTTON_REASON)
|
|
return CallbackResult(ANSWER_DONE, reply)
|
|
if self._deps.mutes is None:
|
|
return CallbackResult(ANSWER_ERROR)
|
|
reply = await create_mute(self._deps.mutes, event.backend, event.sender_id, literal_glob(host),
|
|
literal_glob(name), duration, BUTTON_REASON)
|
|
return CallbackResult(ANSWER_DONE, reply)
|
|
|
|
async def _resolve(self, event: IncomingCallback, parts: tuple[str, ...], approve: bool) -> CallbackResult:
|
|
if self._deps.downtime is None or len(parts) != 1:
|
|
return CallbackResult(ANSWER_ERROR)
|
|
result = await self._deps.downtime.resolve(parts[0], event.backend, event.sender_id, approve)
|
|
return CallbackResult(result.message[:200], plain(result.message), clear_buttons=result.final)
|
|
|