diff --git a/backend/fluksio/api/routes/alerts.py b/backend/fluksio/api/routes/alerts.py index 9853bd5..256ad00 100644 --- a/backend/fluksio/api/routes/alerts.py +++ b/backend/fluksio/api/routes/alerts.py @@ -13,8 +13,9 @@ from fluksio.api.deps import ( get_current_user, ) from fluksio.core.config import settings -from fluksio.flow import webpush +from fluksio.flow import telegram, webpush from fluksio.flow.alerts import Alert, AlertsConfig +from fluksio.flow.secrets import SECRET_REF, SecretNotFound, resolve_params from fluksio.models import Message # Reading or changing where alerts go is a superuser's: a channel setting may @@ -71,6 +72,20 @@ async def save_alerts_config(body: AlertsConfig, controller: FlowControllerDep) detail=f"No channel named {', '.join(repr(m) for m in missing)}", ) + for channel in body.channels: + token = channel.config.get("token") + # A bot token is the bot: it belongs in the secrets store, and this + # file is read by anyone who can read the data directory. + if ( + channel.kind == "telegram" + and token + and not (isinstance(token, dict) and set(token) == {SECRET_REF}) + ): + raise HTTPException( + status_code=422, + detail=f"The bot token of '{channel.name}' has to be a stored secret", + ) + await run_in_threadpool(write_config, body) if controller.alerts is not None: controller.alerts.config = body @@ -107,6 +122,105 @@ async def remove_webpush_subscription(body: Unsubscribe) -> Any: return Message(message="This browser will no longer be notified") +class TelegramLink(BaseModel): + #: Opens the bot in a private chat; pressing Start connects that chat. + url: str + #: Asks which group to add the bot to; the group is connected instead. + group_url: str + expires_in: int + + +class TelegramLinkStatus(BaseModel): + #: A code is out and the engine is listening for it. + waiting: bool + expires_in: int = 0 + #: Why it stopped listening, when it did not run out. + error: str = "" + + +def _store_chat(channel_name: str, chat_id: str, label: str) -> AlertsConfig | None: + """Put a chat that proved itself into the stored channel. Blocking.""" + config = read_config() + channel = next((c for c in config.channels if c.name == channel_name), None) + if channel is None: + # Renamed or removed while the code was out. + return None + channel.config = {**channel.config, "chat_id": chat_id, "label": label} + write_config(config) + return config + + +@router.post( + "/telegram/{channel_name}/link", + response_model=TelegramLink, + dependencies=SUPERUSER, +) +async def start_telegram_link(channel_name: str, controller: FlowControllerDep) -> Any: + """A one-time link that connects whichever chat opens it first. + + Typing a chat id would prove nothing about whose it is. Here the chat + sends the code back through the bot, so it is one somebody opened with a + link the operator handed out. Good for ten minutes, and for one chat. + """ + config = await run_in_threadpool(read_config) + channel = next( + (c for c in config.channels if c.name == channel_name and c.kind == "telegram"), + None, + ) + if channel is None: + raise HTTPException( + status_code=404, detail=f"No telegram channel '{channel_name}'" + ) + try: + token = resolve_params(channel.config.get("token")) + except (SecretNotFound, RuntimeError) as exc: + raise HTTPException(status_code=422, detail=str(exc)) + if not token: + raise HTTPException( + status_code=422, + detail="Store the bot token and save the channel first", + ) + try: + bot = await telegram.call(str(token), "getMe") + except telegram.TelegramError as exc: + raise HTTPException(status_code=502, detail=str(exc)) + + async def linked(name: str, chat_id: str, label: str) -> None: + stored = await run_in_threadpool(_store_chat, name, chat_id, label) + if stored is not None and controller.alerts is not None: + controller.alerts.config = stored + + code = telegram.linker.start(channel_name, str(token), linked) + base = f"https://t.me/{bot['username']}" + return TelegramLink( + url=f"{base}?start={code}", + group_url=f"{base}?startgroup={code}", + expires_in=int(telegram.LINK_TTL_S), + ) + + +@router.get( + "/telegram/{channel_name}/link", + response_model=TelegramLinkStatus, + dependencies=SUPERUSER, +) +async def read_telegram_link(channel_name: str) -> Any: + """Whether a code for this channel is still waiting to be used.""" + waiting, left, error = telegram.linker.status(channel_name) + return TelegramLinkStatus(waiting=waiting, expires_in=left, error=error) + + +@router.delete( + "/telegram/{channel_name}/link", + response_model=Message, + dependencies=SUPERUSER, +) +async def cancel_telegram_link(channel_name: str) -> Any: + """Stop waiting for a chat to open the link.""" + telegram.linker.cancel(channel_name) + return Message(message="Stopped waiting for a chat") + + @router.post("/test/{channel_name}", response_model=Message, dependencies=SUPERUSER) async def test_channel(channel_name: str, controller: FlowControllerDep) -> Any: """Send one alert, to prove the channel works before relying on it.""" diff --git a/backend/fluksio/flow/alerts.py b/backend/fluksio/flow/alerts.py index 9eeac46..4450cba 100644 --- a/backend/fluksio/flow/alerts.py +++ b/backend/fluksio/flow/alerts.py @@ -61,11 +61,13 @@ class Channel(BaseModel): """Somewhere to send an alert.""" name: str - kind: Literal["ntfy", "smtp", "webhook", "dashboard", "webpush"] + kind: Literal["ntfy", "smtp", "webhook", "dashboard", "webpush", "telegram"] enabled: bool = True # ntfy: server + topic + optional token. smtp: to. webhook: url. # dashboard: message — the record a notification widget reads. # webpush: nothing — it goes to whichever browsers subscribed here. + # telegram: token (always a secret reference) + chat_id and label, which + # connecting a chat fills in. One channel is one chat. config: dict[str, Any] = Field(default_factory=dict) @@ -414,6 +416,8 @@ class AlertManager: await self._send_dashboard(config, alert) elif channel.kind == "webpush": await self._send_webpush(alert) + elif channel.kind == "telegram": + await self._send_telegram(config, alert) else: await self._send_email(config, alert) except Exception as exc: @@ -466,6 +470,23 @@ class AlertManager: if await webpush.send_to_all(alert.model_dump()) == 0: raise ValueError("No browsers are subscribed") + async def _send_telegram(self, config: dict[str, Any], alert: Alert) -> None: + from fluksio.flow import telegram + + token = config.get("token") + if not token: + raise ValueError("a telegram channel needs a bot token") + chat_id = config.get("chat_id") + if not chat_id: + raise ValueError("this telegram channel is not connected to a chat yet") + await telegram.send_message( + str(token), + str(chat_id), + alert.title, + alert.body, + quiet=alert.severity == "warning", + ) + async def _send_webhook(self, config: dict[str, Any], alert: Alert) -> None: url = config.get("url") if not url: diff --git a/backend/fluksio/flow/telegram.py b/backend/fluksio/flow/telegram.py new file mode 100644 index 0000000..7ebc4fe --- /dev/null +++ b/backend/fluksio/flow/telegram.py @@ -0,0 +1,303 @@ +"""Telegram: an alert reaching a chat. + +The Bot API is plain HTTPS and JSON, so ``httpx`` is all it needs. Two things +about it shape this module. + +The token is part of every URL (``/bot/sendMessage``), so anything that +prints a URL prints the token. Errors are therefore built from Telegram's own +description and never from the request, and ``redact`` is what the log filter +in ``main.py`` applies to the rest. + +A bot cannot message someone who has not started it, and anyone can start a +public bot. Which chat belongs to whom is settled by a one-time code: the +operator asks for a link, opens it in Telegram, and presses Start, which sends +the code back as ``/start ``. Updates are fetched with ``getUpdates`` +only while such a code is waiting, so there is no standing connection and no +inbound port: nothing reaches this instance that it did not ask for. Anything +other than a live code is ignored without an answer. +""" + +from __future__ import annotations + +import asyncio +import html +import logging +import re +import secrets +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Any + +import httpx + +logger = logging.getLogger(__name__) + +API = "https://api.telegram.org" +#: A message is 1 to 4096 characters once its markup is parsed. Under it, so +#: the title and the ellipsis always fit. +MAX_TEXT = 4000 +#: Longest a rate-limited send waits for the "retry after" Telegram names. +MAX_RETRY_S = 30.0 +#: How long a code stays good. +LINK_TTL_S = 600.0 +#: How long one ``getUpdates`` waits for news. +POLL_S = 25 + +_TOKEN_IN_URL = re.compile(r"(api\.telegram\.org/bot)[^/\s\"']+") +#: What Start sends: ``/start `` in a private chat, ``/start@bot `` +#: in a group. +_START = re.compile(r"^/start(?:@\w+)?\s+([A-Za-z0-9_-]{1,64})\s*$") + + +def redact(text: str) -> str: + """Hide the bot token in anything that quotes a Bot API URL.""" + return _TOKEN_IN_URL.sub(r"\1***", text) + + +class TelegramError(Exception): + """The Bot API refused, or could not be reached. Never quotes a URL.""" + + def __init__( + self, + message: str, + code: int = 0, + retry_after: float = 0.0, + ) -> None: + super().__init__(message) + self.code = code + self.retry_after = retry_after + + +def _client(timeout: float) -> httpx.AsyncClient: + return httpx.AsyncClient(timeout=timeout) + + +def _explain(code: int, description: str) -> str: + """What the operator should do about it, for the errors that have an answer.""" + lowered = description.lower() + if code == 401: + return "Telegram rejected the bot token. Check it with @BotFather." + if "blocked" in lowered or "kicked" in lowered or "deactivated" in lowered: + return ( + "The bot was blocked or removed from the chat. Connect the channel again." + ) + if "chat not found" in lowered: + return "Telegram does not know this chat. Connect the channel again." + if "upgraded to a supergroup" in lowered: + return "The group became a supergroup, so its id changed. Connect the channel again." + if code == 409: + return ( + "Another program is fetching this bot's updates, or a webhook is set. " + "Use a bot of its own for this instance." + ) + return f"Telegram refused: {description or code}" + + +async def call( + token: str, + method: str, + payload: dict[str, Any] | None = None, + timeout: float = 10.0, +) -> Any: + """One Bot API method. Returns ``result``; raises ``TelegramError``.""" + try: + async with _client(timeout) as client: + response = await client.post( + f"{API}/bot{token}/{method}", json=payload or {} + ) + except (httpx.HTTPError, httpx.InvalidURL) as exc: + raise TelegramError( + redact(f"Could not reach Telegram ({type(exc).__name__}: {exc})") + ) from None + try: + body = response.json() + except ValueError: + body = {} + if response.is_success and body.get("ok"): + return body.get("result") + code = int(body.get("error_code") or response.status_code) + parameters = body.get("parameters") or {} + raise TelegramError( + _explain(code, str(body.get("description") or "")), + code=code, + retry_after=float(parameters.get("retry_after") or 0), + ) + + +def format_message(title: str, body: str) -> str: + """HTML, because it needs only three characters escaped. Always short enough.""" + + def build(text: str) -> str: + return f"{html.escape(title)}\n{html.escape(text)}" + + if len(build(body)) <= MAX_TEXT: + return build(body) + keep = body + while keep and len(build(keep + "…")) > MAX_TEXT: + keep = keep[: len(keep) * 4 // 5] + return build(keep + "…") + + +async def send_message( + token: str, chat_id: str, title: str, body: str, quiet: bool = False +) -> None: + """Send one alert to a chat. A 429 is waited out once, if it is short.""" + payload = { + "chat_id": chat_id, + "text": format_message(title, body), + "parse_mode": "HTML", + "link_preview_options": {"is_disabled": True}, + # A warning arrives without a sound; an error does not. + "disable_notification": quiet, + } + try: + await call(token, "sendMessage", payload) + except TelegramError as exc: + if exc.code != 429: + raise + if not 0 < exc.retry_after <= MAX_RETRY_S: + raise TelegramError( + f"Telegram is rate limiting this bot. Try again in {exc.retry_after:.0f} s.", + code=429, + ) from None + await asyncio.sleep(exc.retry_after) + await call(token, "sendMessage", payload) + + +# ----------------------------------------------------------------------------- +# Connecting a chat +# ----------------------------------------------------------------------------- + +#: Called with ``(channel, chat_id, label)`` once a chat has proved itself. +Linked = Callable[[str, str, str], Awaitable[None]] + + +@dataclass +class _Pending: + channel: str + token: str + expires: float + on_linked: Linked + error: str = "" + + +class Linker: + """The codes waiting for a ``/start``, and the poll that listens for it.""" + + def __init__(self) -> None: + self._pending: dict[str, _Pending] = {} + self._tasks: dict[str, asyncio.Task[None]] = {} + + def start( + self, + channel: str, + token: str, + on_linked: Linked, + ttl: float = LINK_TTL_S, + ) -> str: + """A new code for *channel*, replacing any it had. Begins listening.""" + self.cancel(channel) + code = secrets.token_urlsafe(16) + self._pending[code] = _Pending( + channel, token, time.monotonic() + ttl, on_linked + ) + # One poll per bot: two would each be told the other is in the way. + running = self._tasks.get(token) + if running is None or running.done(): + task = asyncio.create_task(self._poll(token), name="telegram-link") + self._tasks[token] = task + task.add_done_callback( + lambda t: self._tasks.pop(token) + if self._tasks.get(token) is t + else None + ) + return code + + def cancel(self, channel: str) -> None: + for code in [c for c, p in self._pending.items() if p.channel == channel]: + del self._pending[code] + + def status(self, channel: str) -> tuple[bool, int, str]: + """``(waiting, seconds left, why it stopped)`` for *channel*.""" + self._expire() + for pending in self._pending.values(): + if pending.channel == channel: + left = max(0, int(pending.expires - time.monotonic())) + return pending.error == "", left, pending.error + return False, 0, "" + + def _expire(self) -> None: + now = time.monotonic() + for code in [c for c, p in self._pending.items() if p.expires <= now]: + del self._pending[code] + + def _waiting(self, token: str) -> bool: + self._expire() + return any(p.token == token and not p.error for p in self._pending.values()) + + async def _poll(self, token: str) -> None: + offset: int | None = None + while self._waiting(token): + payload: dict[str, Any] = { + "timeout": POLL_S, + "allowed_updates": ["message"], + } + if offset is not None: + payload["offset"] = offset + try: + updates = await call(token, "getUpdates", payload, timeout=POLL_S + 15) + except TelegramError as exc: + if exc.code in (401, 409): + # Nothing a retry fixes; say so where the operator looks. + for pending in self._pending.values(): + if pending.token == token: + pending.error = str(exc) + return + logger.warning("Telegram link poll failed: %s", exc) + await asyncio.sleep(5) + continue + for update in updates or []: + offset = int(update["update_id"]) + 1 + try: + await self._take(token, update) + except Exception: + logger.exception("Could not handle a Telegram update") + + async def _take(self, token: str, update: dict[str, Any]) -> None: + message = update.get("message") or {} + match = _START.match(str(message.get("text") or "")) + if match is None: + return + self._expire() + pending = self._pending.get(match.group(1)) + if pending is None or pending.token != token: + return + del self._pending[match.group(1)] + chat = message.get("chat") or {} + chat_id = str(chat.get("id") or "") + if not chat_id: + return + label = ( + chat.get("title") + or " ".join( + part for part in (chat.get("first_name"), chat.get("last_name")) if part + ) + or chat.get("username") + or chat_id + ) + await pending.on_linked(pending.channel, chat_id, str(label)) + try: + await call( + token, + "sendMessage", + { + "chat_id": chat_id, + "text": "Connected. Alerts from this Fluksio will arrive here.", + }, + ) + except TelegramError as exc: + logger.warning("Could not confirm a Telegram link: %s", exc) + + +linker = Linker() diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index c9ba188..f597f58 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -12,6 +12,7 @@ from contextlib import AbstractAsyncContextManager, asynccontextmanager from pathlib import Path from typing import Any +import httpx from fastapi import FastAPI, Request from fastapi.concurrency import run_in_threadpool from fastapi.responses import JSONResponse @@ -29,7 +30,7 @@ from fluksio.core import security from fluksio.core.config import settings from fluksio.core.db import engine as db_engine from fluksio.core.db import prepare -from fluksio.flow import logs, modules +from fluksio.flow import logs, modules, telegram from fluksio.flow.alert_ledger import AlertLedger from fluksio.flow.alerts import AlertManager from fluksio.flow.artifacts import ArtifactStore, VolatileStore @@ -74,7 +75,9 @@ _SECRET_IN_HOOK_PATH = re.compile(r"(/hooks/[^/\s?]+/(?:[^/\s?]*/)*)[^/\s?]+") def _redact(text: str) -> str: - return _SECRET_IN_HOOK_PATH.sub(r"\1***", _CREDENTIAL_IN_QUERY.sub(r"\1***", text)) + return telegram.redact( + _SECRET_IN_HOOK_PATH.sub(r"\1***", _CREDENTIAL_IN_QUERY.sub(r"\1***", text)) + ) class _RedactTokens(logging.Filter): @@ -87,15 +90,19 @@ class _RedactTokens(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: if isinstance(record.args, tuple): + # httpx logs its URL as an `httpx.URL`, not a string. record.args = tuple( - _redact(arg) if isinstance(arg, str) else arg for arg in record.args + _redact(str(arg)) if isinstance(arg, str | httpx.URL) else arg + for arg in record.args ) return True # The handshake lines go to `uvicorn.error`, the request lines to `uvicorn.access`, # and the tunnel connector names the path of a call it could not proxy. -for _name in ("uvicorn.access", "uvicorn.error", "fluksio.cloud.connector"): +# httpx logs every request it makes at INFO, with the URL, and a Telegram bot's +# token is part of that URL. +for _name in ("uvicorn.access", "uvicorn.error", "fluksio.cloud.connector", "httpx"): logging.getLogger(_name).addFilter(_RedactTokens()) diff --git a/backend/tests/api/routes/test_alerts.py b/backend/tests/api/routes/test_alerts.py index 74dd1e6..7aa624f 100644 --- a/backend/tests/api/routes/test_alerts.py +++ b/backend/tests/api/routes/test_alerts.py @@ -140,3 +140,137 @@ def test_any_account_can_subscribe_its_own_browser( assert response.status_code == 200 assert response.json()["key"] + + +def _telegram_config(token: Any) -> dict[str, Any]: + return { + "channels": [ + { + "name": "garage", + "kind": "telegram", + "config": {"token": token} if token else {}, + } + ], + "rules": [], + } + + +def test_a_bot_token_has_to_be_a_stored_secret( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + plain = client.put( + f"{PREFIX}/config", + headers=superuser_token_headers, + json=_telegram_config("123456:literal-token"), + ) + stored = client.put( + f"{PREFIX}/config", + headers=superuser_token_headers, + json=_telegram_config({"$secret": "telegram-bot"}), + ) + + assert plain.status_code == 422 + assert "stored secret" in plain.json()["detail"] + assert stored.status_code == 200 + + +def test_a_chat_connects_through_the_link_and_the_token_is_never_returned( + client: TestClient, + superuser_token_headers: dict[str, str], + normal_user_token_headers: dict[str, str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + import json + import time + + import httpx + + from fluksio.flow import telegram + + token = "123456:SECRET-token" + client.put( + f"{settings.API_V1_STR}/secrets/telegram-bot", + headers=superuser_token_headers, + json={"value": token}, + ) + client.put( + f"{PREFIX}/config", + headers=superuser_token_headers, + json=_telegram_config({"$secret": "telegram-bot"}), + ) + + def bot(request: httpx.Request) -> httpx.Response: + method = request.url.path.rsplit("/", 1)[1] + assert request.url.path.startswith(f"/bot{token}/") + result: Any = True + if method == "getMe": + result = {"username": "fluksio_bot"} + elif method == "getUpdates": + time.sleep(0.05) + codes = [c for c, p in telegram.linker._pending.items() if p.token == token] + result = ( + [ + { + "update_id": 1, + "message": { + "text": f"/start {codes[0]}", + "chat": {"id": 42, "first_name": "Anna"}, + }, + } + ] + if codes + else [] + ) + return httpx.Response(200, content=json.dumps({"ok": True, "result": result})) + + transport = httpx.MockTransport(bot) + monkeypatch.setattr( + telegram, + "_client", + lambda timeout: httpx.AsyncClient(transport=transport, timeout=timeout), + ) + + link_url = f"{PREFIX}/telegram/garage/link" + assert client.post(link_url, headers=normal_user_token_headers).status_code == 403 + response = client.post(link_url, headers=superuser_token_headers) + + assert response.status_code == 200 + body = response.json() + assert body["url"].startswith("https://t.me/fluksio_bot?start=") + assert body["group_url"].startswith("https://t.me/fluksio_bot?startgroup=") + assert token not in response.text + + channel: dict[str, Any] = {} + for _ in range(100): + config = client.get(f"{PREFIX}/config", headers=superuser_token_headers) + channel = config.json()["channels"][0] + if channel["config"].get("chat_id"): + break + time.sleep(0.05) + + assert channel["config"]["chat_id"] == "42" + assert channel["config"]["label"] == "Anna" + assert channel["config"]["token"] == {"$secret": "telegram-bot"} + assert token not in config.text + status = client.get(link_url, headers=superuser_token_headers).json() + assert status["waiting"] is False + + +def test_a_link_needs_a_stored_token( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + client.put( + f"{PREFIX}/config", + headers=superuser_token_headers, + json=_telegram_config(None), + ) + + response = client.post( + f"{PREFIX}/telegram/garage/link", headers=superuser_token_headers + ) + unknown = client.post( + f"{PREFIX}/telegram/nowhere/link", headers=superuser_token_headers + ) + + assert response.status_code == 422 + assert unknown.status_code == 404 diff --git a/backend/tests/flow/test_telegram.py b/backend/tests/flow/test_telegram.py new file mode 100644 index 0000000..12dd394 --- /dev/null +++ b/backend/tests/flow/test_telegram.py @@ -0,0 +1,276 @@ +"""Telegram: a fake Bot API stands in for api.telegram.org.""" + +import asyncio +import json +import logging +from typing import Any + +import httpx +import pytest + +import fluksio.main # noqa: F401 - installs the log filter +from fluksio.flow import telegram +from fluksio.flow.alerts import Alert, AlertManager, Channel +from fluksio.flow.events import EventBus + +TOKEN = "123456:SECRET-token" + + +class FakeBot: + """Answers the Bot API, and remembers what it was asked.""" + + def __init__(self, monkeypatch: pytest.MonkeyPatch) -> None: + self.calls: list[tuple[str, dict[str, Any]]] = [] + #: Per method: a list of ``(status, body)`` answers, used in order. + self.answers: dict[str, list[tuple[int, dict[str, Any]]]] = {} + transport = httpx.MockTransport(self._handle) + monkeypatch.setattr( + telegram, + "_client", + lambda timeout: httpx.AsyncClient(transport=transport, timeout=timeout), + ) + + def _handle(self, request: httpx.Request) -> httpx.Response: + assert request.url.path.startswith(f"/bot{TOKEN}/") + method = request.url.path.rsplit("/", 1)[1] + self.calls.append((method, json.loads(request.content or b"{}"))) + queued = self.answers.get(method) + if queued: + status, body = queued.pop(0) if len(queued) > 1 else queued[0] + return httpx.Response(status, json=body) + return httpx.Response(200, json={"ok": True, "result": True}) + + def failing(self, method: str, status: int, description: str, **extra: Any) -> None: + body = {"ok": False, "error_code": status, "description": description} + if extra: + body["parameters"] = extra + self.answers[method] = [(status, body)] + + def sent(self) -> list[dict[str, Any]]: + return [payload for method, payload in self.calls if method == "sendMessage"] + + +@pytest.fixture +def bot(monkeypatch: pytest.MonkeyPatch) -> FakeBot: + return FakeBot(monkeypatch) + + +def test_an_alert_is_sent_escaped_and_quiet_when_it_is_a_warning(bot: FakeBot) -> None: + asyncio.run( + telegram.send_message(TOKEN, "42", " failed", "a & b ", quiet=True) + ) + + (message,) = bot.sent() + assert message["chat_id"] == "42" + assert message["parse_mode"] == "HTML" + assert message["text"] == "<pump> failed\na & b <c>" + assert message["disable_notification"] is True + + +def test_a_long_body_is_cut_to_fit_a_message(bot: FakeBot) -> None: + asyncio.run(telegram.send_message(TOKEN, "42", "t", "<" * 10_000)) + + (message,) = bot.sent() + assert len(message["text"]) < 4096 + assert message["text"].endswith("…") + + +@pytest.mark.parametrize( + ("status", "description", "expected"), + [ + (403, "Forbidden: bot was blocked by the user", "blocked or removed"), + (400, "Bad Request: chat not found", "does not know this chat"), + (401, "Unauthorized", "rejected the bot token"), + ], +) +def test_a_refusal_says_what_to_do_and_never_quotes_the_url( + bot: FakeBot, status: int, description: str, expected: str +) -> None: + bot.failing("sendMessage", status, description) + + with pytest.raises(telegram.TelegramError) as raised: + asyncio.run(telegram.send_message(TOKEN, "42", "t", "b")) + + assert expected in str(raised.value) + assert "SECRET" not in str(raised.value) + + +def test_a_short_rate_limit_is_waited_out_once(bot: FakeBot) -> None: + bot.answers["sendMessage"] = [ + (429, {"ok": False, "error_code": 429, "parameters": {"retry_after": 0.01}}), + (200, {"ok": True, "result": True}), + ] + + asyncio.run(telegram.send_message(TOKEN, "42", "t", "b")) + + assert len(bot.sent()) == 2 + + +def test_a_long_rate_limit_is_reported_instead(bot: FakeBot) -> None: + bot.failing("sendMessage", 429, "Too Many Requests", retry_after=120) + + with pytest.raises(telegram.TelegramError, match="rate limiting"): + asyncio.run(telegram.send_message(TOKEN, "42", "t", "b")) + + assert len(bot.sent()) == 1 + + +def test_a_network_error_does_not_carry_the_token( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def refuse(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError(f"cannot reach {request.url}") + + transport = httpx.MockTransport(refuse) + monkeypatch.setattr( + telegram, + "_client", + lambda timeout: httpx.AsyncClient(transport=transport, timeout=timeout), + ) + + with pytest.raises(telegram.TelegramError) as raised: + asyncio.run(telegram.call(TOKEN, "getMe")) + + assert "SECRET" not in str(raised.value) + assert "api.telegram.org/bot***" in str(raised.value) + + +def test_the_token_stays_out_of_httpx_log_lines( + bot: FakeBot, caplog: pytest.LogCaptureFixture +) -> None: + with caplog.at_level(logging.INFO, logger="httpx"): + asyncio.run(telegram.call(TOKEN, "getMe")) + + assert "api.telegram.org/bot***/getMe" in caplog.text + assert "SECRET" not in caplog.text + + +def test_an_alert_channel_needs_a_token_and_a_chat() -> None: + alerts = AlertManager(EventBus()) + alert = Alert(title="t", body="b") + + async def send(config: dict[str, Any]) -> None: + await alerts.send( + Channel(name="phone", kind="telegram", config=config), + alert, + raise_on_error=True, + ) + + with pytest.raises(ValueError, match="bot token"): + asyncio.run(send({})) + with pytest.raises(ValueError, match="not connected"): + asyncio.run(send({"token": TOKEN})) + + +# ----------------------------------------------------------------------------- +# Connecting a chat +# ----------------------------------------------------------------------------- + + +def a_start(update_id: int, text: str, **chat: Any) -> dict[str, Any]: + return {"update_id": update_id, "message": {"text": text, "chat": chat}} + + +def link( + bot: FakeBot, updates: list[dict[str, Any]] +) -> tuple[telegram.Linker, list[tuple[str, str, str]], str]: + """Open a code, and have the bot's first poll answer with *updates*. + + ``{code}`` in an update's text is replaced by the code that was issued. + """ + linker = telegram.Linker() + linked: list[tuple[str, str, str]] = [] + + async def on_linked(channel: str, chat_id: str, label: str) -> None: + linked.append((channel, chat_id, label)) + + code = linker.start("garage", TOKEN, on_linked) + filled = json.loads(json.dumps(updates).replace("{code}", code)) + bot.answers["getUpdates"] = [ + (200, {"ok": True, "result": filled}), + (200, {"ok": True, "result": []}), + ] + return linker, linked, code + + +def test_the_chat_that_sends_the_code_is_connected_and_nobody_else( + bot: FakeBot, +) -> None: + async def scenario() -> tuple[telegram.Linker, list[tuple[str, str, str]]]: + linker, linked, _ = link( + bot, + [ + a_start(1, "hello", id=7, first_name="Stranger"), + a_start(2, "/start not-a-live-code", id=8, first_name="Guesser"), + a_start(3, "/start {code}", id=42, first_name="Anna", last_name="B"), + ], + ) + await asyncio.wait_for(asyncio.gather(*linker._tasks.values()), 5) + return linker, linked + + linker, linked = asyncio.run(scenario()) + + assert linked == [("garage", "42", "Anna B")] + # Only the connected chat is answered; the others get no reply at all. + assert [m["chat_id"] for m in bot.sent()] == ["42"] + assert linker.status("garage") == (False, 0, "") + + +def test_a_group_connects_through_start_at_bot(bot: FakeBot) -> None: + async def scenario() -> list[tuple[str, str, str]]: + linker, linked, _ = link( + bot, [a_start(1, "/start@fluksio_bot {code}", id=-100, title="Garage")] + ) + await asyncio.wait_for(asyncio.gather(*linker._tasks.values()), 5) + return linked + + assert asyncio.run(scenario()) == [("garage", "-100", "Garage")] + + +def test_a_code_works_once(bot: FakeBot) -> None: + async def scenario() -> list[tuple[str, str, str]]: + linker, linked, _ = link( + bot, + [ + a_start(1, "/start {code}", id=42, first_name="Anna"), + a_start(2, "/start {code}", id=43, first_name="Ben"), + ], + ) + await asyncio.wait_for(asyncio.gather(*linker._tasks.values()), 5) + return linked + + assert [chat for _, chat, _ in asyncio.run(scenario())] == ["42"] + + +def test_an_expired_code_is_not_listened_for(bot: FakeBot) -> None: + async def scenario() -> bool: + linker = telegram.Linker() + + async def on_linked(channel: str, chat_id: str, label: str) -> None: + raise AssertionError("expired") + + linker.start("garage", TOKEN, on_linked, ttl=0.0) + await asyncio.gather(*linker._tasks.values()) + return linker.status("garage")[0] + + assert asyncio.run(scenario()) is False + assert not [call for call in bot.calls if call[0] == "getUpdates"] + + +def test_a_webhook_or_second_poller_is_reported_not_retried(bot: FakeBot) -> None: + bot.failing("getUpdates", 409, "Conflict: terminated by other getUpdates request") + + async def scenario() -> tuple[bool, int, str]: + linker = telegram.Linker() + + async def on_linked(channel: str, chat_id: str, label: str) -> None: + pass + + linker.start("garage", TOKEN, on_linked) + await asyncio.wait_for(asyncio.gather(*linker._tasks.values()), 5) + return linker.status("garage") + + waiting, _, error = asyncio.run(scenario()) + + assert waiting is False + assert "bot of its own" in error diff --git a/docs/code/api.md b/docs/code/api.md index 585df6b..3519742 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -220,6 +220,7 @@ reclaiming space sooner. | `POST` | `/modules/apply` | replace the manifest with `{"requirements": "numpy>=2\n"}` | | `GET`, `PUT` | `/alerts/config` | read or update alert configuration | | `POST` | `/alerts/test/{channel}` | test a channel | +| `POST`, `GET`, `DELETE` | `/alerts/telegram/{channel}/link` | start, read or cancel connecting a chat to a telegram channel; the start returns the `t.me` links and never the bot token | | `GET` | `/alerts/webpush/key` | the key a browser subscribes against; any account | | `POST` | `/alerts/webpush/subscriptions`, `/alerts/webpush/unsubscribe` | add or remove the calling browser; any account | diff --git a/docs/interface/operations.md b/docs/interface/operations.md index e4cd7b6..c93d1bf 100644 --- a/docs/interface/operations.md +++ b/docs/interface/operations.md @@ -63,10 +63,36 @@ subscribes its own browser to a **webpush** channel. | **Webhook** | URL that receives a POST | | **Dashboard** | declared message read by a Notification widget | | **webpush** | browsers subscribed from the Alerts page | +| **Telegram** | one chat or group, connected from the Alerts page through a bot of your own | Channel settings may use `{"$secret": "name"}` references. Browser push also needs notification permission and a secure browser context. +### Telegram + +Create a bot with @BotFather and copy its token. Add a **telegram** channel, +paste the token under **Bot token**, press **Store token**, then **Save**. The +token goes to the secrets store and the channel keeps only a reference, which +is the one form the API accepts. + +Press **Connect**, open the link in Telegram on the device that should get the +alerts and press Start. **Add to a group** connects a group instead, and +everyone in it reads the alerts. The link works once and runs out after ten +minutes. A chat id is never typed in: the chat that sends the link's code back +is the one connected. + +One channel is one chat. To send different events to different people, make +a channel for each and choose them in the rules. A warning arrives without a +sound. If the chat blocks or removes the bot, **Test** and the log say so; +press **Reconnect**. + +The engine asks Telegram for updates only while a link is waiting, so it needs +no public address, and anything else sent to the bot is ignored without a +reply. Telegram lets one program fetch a bot's updates at a time, so give each +instance a bot of its own. Starting from a group sends `/start@bot code`; +whether a group in Telegram's default privacy mode delivers it has not been +tried against a live bot, so press **Test** after connecting one. + ### Rules A rule selects events and destination channels. Leaving all events unchecked diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 0c4bd23..8c9fe61 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -485,7 +485,7 @@ export const ChannelSchema = { }, kind: { type: 'string', - enum: ['ntfy', 'smtp', 'webhook', 'dashboard', 'webpush'], + enum: ['ntfy', 'smtp', 'webhook', 'dashboard', 'webpush', 'telegram'], title: 'Kind' }, enabled: { @@ -4440,6 +4440,48 @@ export const TargetResourcesSchema = { description: 'One machine: this engine, or a worker attached to it.' } as const; +export const TelegramLinkSchema = { + properties: { + url: { + type: 'string', + title: 'Url' + }, + group_url: { + type: 'string', + title: 'Group Url' + }, + expires_in: { + type: 'integer', + title: 'Expires In' + } + }, + type: 'object', + required: ['url', 'group_url', 'expires_in'], + title: 'TelegramLink' +} as const; + +export const TelegramLinkStatusSchema = { + properties: { + waiting: { + type: 'boolean', + title: 'Waiting' + }, + expires_in: { + type: 'integer', + title: 'Expires In', + default: 0 + }, + error: { + type: 'string', + title: 'Error', + default: '' + } + }, + type: 'object', + required: ['waiting'], + title: 'TelegramLinkStatus' +} as const; + export const TokenSchema = { properties: { access_token: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 73ab50a..9832b4e 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesReadIceData, DevicesReadIceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsTrainNodeData, FlowsTrainNodeResponse, FlowsReadTrainingData, FlowsReadTrainingResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsStartTelegramLinkData, AlertsStartTelegramLinkResponse, AlertsReadTelegramLinkData, AlertsReadTelegramLinkResponse, AlertsCancelTelegramLinkData, AlertsCancelTelegramLinkResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesReadIceData, DevicesReadIceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsTrainNodeData, FlowsTrainNodeResponse, FlowsReadTrainingData, FlowsReadTrainingResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AiService { /** @@ -238,6 +238,73 @@ export class AlertsService { }); } + /** + * Start Telegram Link + * A one-time link that connects whichever chat opens it first. + * + * Typing a chat id would prove nothing about whose it is. Here the chat + * sends the code back through the bot, so it is one somebody opened with a + * link the operator handed out. Good for ten minutes, and for one chat. + * @param data The data for the request. + * @param data.channelName + * @returns TelegramLink Successful Response + * @throws ApiError + */ + public static startTelegramLink(data: AlertsStartTelegramLinkData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/alerts/telegram/{channel_name}/link', + path: { + channel_name: data.channelName + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Read Telegram Link + * Whether a code for this channel is still waiting to be used. + * @param data The data for the request. + * @param data.channelName + * @returns TelegramLinkStatus Successful Response + * @throws ApiError + */ + public static readTelegramLink(data: AlertsReadTelegramLinkData): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/alerts/telegram/{channel_name}/link', + path: { + channel_name: data.channelName + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Cancel Telegram Link + * Stop waiting for a chat to open the link. + * @param data The data for the request. + * @param data.channelName + * @returns Message Successful Response + * @throws ApiError + */ + public static cancelTelegramLink(data: AlertsCancelTelegramLinkData): CancelablePromise { + return __request(OpenAPI, { + method: 'DELETE', + url: '/api/v1/alerts/telegram/{channel_name}/link', + path: { + channel_name: data.channelName + }, + errors: { + 422: 'Validation Error' + } + }); + } + /** * Test Channel * Send one alert, to prove the channel works before relying on it. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 894d116..bccee88 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -123,14 +123,14 @@ export type BrainNode = { */ export type Channel = { name: string; - kind: 'ntfy' | 'smtp' | 'webhook' | 'dashboard' | 'webpush'; + kind: 'ntfy' | 'smtp' | 'webhook' | 'dashboard' | 'webpush' | 'telegram'; enabled?: boolean; config?: { [key: string]: unknown; }; }; -export type kind = 'ntfy' | 'smtp' | 'webhook' | 'dashboard' | 'webpush'; +export type kind = 'ntfy' | 'smtp' | 'webhook' | 'dashboard' | 'webpush' | 'telegram'; export type ChatItem = { role: string; @@ -1454,6 +1454,18 @@ export type TargetResources = { in_flight?: number; }; +export type TelegramLink = { + url: string; + group_url: string; + expires_in: number; +}; + +export type TelegramLinkStatus = { + waiting: boolean; + expires_in?: number; + error?: string; +}; + export type Token = { access_token: string; token_type?: string; @@ -1740,6 +1752,24 @@ export type AlertsRemoveWebpushSubscriptionData = { export type AlertsRemoveWebpushSubscriptionResponse = (Message); +export type AlertsStartTelegramLinkData = { + channelName: string; +}; + +export type AlertsStartTelegramLinkResponse = (TelegramLink); + +export type AlertsReadTelegramLinkData = { + channelName: string; +}; + +export type AlertsReadTelegramLinkResponse = (TelegramLinkStatus); + +export type AlertsCancelTelegramLinkData = { + channelName: string; +}; + +export type AlertsCancelTelegramLinkResponse = (Message); + export type AlertsTestChannelData = { channelName: string; }; diff --git a/frontend/src/components/Alerts/TelegramConnect.tsx b/frontend/src/components/Alerts/TelegramConnect.tsx new file mode 100644 index 0000000..b013f81 --- /dev/null +++ b/frontend/src/components/Alerts/TelegramConnect.tsx @@ -0,0 +1,268 @@ +import { useMutation, useQuery } from "@tanstack/react-query" +import { Check, Copy, ExternalLink, Send, Unplug, Users } from "lucide-react" +import { useEffect, useRef, useState } from "react" + +import { + type AlertsConfig, + AlertsService, + type Channel, + SecretsService, + type TelegramLink, +} from "@/client" +import { InfoTip } from "@/components/Flow/SidePanel" +import { Button } from "@/components/ui/button" +import { Input } from "@/components/ui/input" +import { Label } from "@/components/ui/label" +import { useCopyToClipboard } from "@/hooks/useCopyToClipboard" +import useCustomToast from "@/hooks/useCustomToast" +import { handleError } from "@/utils" + +/** Where a token entered here is kept, unless the channel already names one. */ +const DEFAULT_SECRET = "telegram-bot" + +type Config = Record + +/** + * The two things a Telegram channel needs beyond a name: the bot's token, and + * the chat it writes to. + * + * The token goes to the secrets store and the channel keeps only a reference, + * so it is never shown again. The chat is never typed: the engine hands out a + * link, and the chat that opens it and presses Start is the one connected. + * That needs the stored channel, so Connect waits for a saved form. + */ +export function TelegramConnect({ + channel, + dirty, + onPatch, + onConnected, +}: { + channel: Channel + /** Unsaved changes anywhere in the form. */ + dirty: boolean + /** Merge into this channel's settings, in the draft. */ + onPatch: (config: Config) => void + /** The engine connected a chat: the stored config, to adopt. */ + onConnected: (stored: AlertsConfig) => void +}) { + const { showSuccessToast, showErrorToast } = useCustomToast() + const [copied, copy] = useCopyToClipboard() + const config = (channel.config ?? {}) as Config + const reference = config.token as { $secret?: string } | undefined + const secretName = reference?.$secret + const chatId = config.chat_id ? String(config.chat_id) : "" + const label = config.label ? String(config.label) : chatId + + const [replacing, setReplacing] = useState(false) + const [token, setToken] = useState("") + const [link, setLink] = useState(null) + + const storeToken = useMutation({ + mutationFn: (value: string) => + SecretsService.saveSecret({ + name: secretName ?? DEFAULT_SECRET, + requestBody: { value }, + }), + onSuccess: () => { + setToken("") + setReplacing(false) + onPatch({ token: { $secret: secretName ?? DEFAULT_SECRET } }) + showSuccessToast("Bot token stored. Save to use it") + }, + onError: handleError.bind(showErrorToast), + }) + + const connect = useMutation({ + mutationFn: () => + AlertsService.startTelegramLink({ channelName: channel.name }), + onSuccess: setLink, + onError: handleError.bind(showErrorToast), + }) + + const cancel = useMutation({ + mutationFn: () => + AlertsService.cancelTelegramLink({ channelName: channel.name }), + onSettled: () => setLink(null), + }) + + // The engine listens for the chat only while a code is out. Once it stops, + // the stored channel says whether a chat answered. + const status = useQuery({ + queryKey: ["alerts", "telegram-link", channel.name], + queryFn: () => + AlertsService.readTelegramLink({ channelName: channel.name }), + enabled: link !== null, + refetchInterval: 2000, + }) + const finished = link !== null && status.data?.waiting === false + const failure = status.data?.error ?? "" + const handled = useRef(null) + useEffect(() => { + if (!finished || handled.current === link) return + handled.current = link + setLink(null) + AlertsService.readAlertsConfig().then((stored) => { + const now = stored.channels?.find((c) => c.name === channel.name)?.config + if (now?.chat_id && String(now.chat_id) !== chatId) { + onConnected(stored) + showSuccessToast(`Connected to ${String(now.label ?? now.chat_id)}`) + } else { + showErrorToast(failure || "The link ran out. Connect again") + } + }) + }, [ + finished, + link, + failure, + channel.name, + chatId, + onConnected, + showSuccessToast, + showErrorToast, + ]) + + const needsSave = dirty + ? "Save first. The engine connects the stored channel" + : undefined + const showTokenField = !secretName || replacing + + return ( +
+
+
+ + + Create a bot with @BotFather in Telegram and paste its token. It is + stored as a secret and not shown again. Use one bot per instance. + +
+ {showTokenField ? ( +
+ setToken(event.target.value)} + /> + + {replacing ? ( + + ) : null} +
+ ) : ( +
+ + Stored as secret {secretName} + + +
+ )} +
+ +
+ + {link ? ( +
+ + Open the link in Telegram and press Start. + +
+ + + + + + Open the link on the device that should get the alerts. Add to a + group connects that group instead, and everyone in it reads the + alerts. The link works once and runs out after ten minutes. + +
+
+ ) : ( +
+ + {chatId + ? `Connected to ${label} (${chatId.startsWith("-") ? "group" : "private chat"})` + : "Not connected"} + + + {chatId ? ( + + ) : null} +
+ )} +
+
+ ) +} diff --git a/frontend/src/routes/_layout/alerts.tsx b/frontend/src/routes/_layout/alerts.tsx index 544dc6f..27b5c8b 100644 --- a/frontend/src/routes/_layout/alerts.tsx +++ b/frontend/src/routes/_layout/alerts.tsx @@ -1,9 +1,10 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query" import { createFileRoute } from "@tanstack/react-router" import { Bell, BellOff, Plus, Send, Trash2 } from "lucide-react" -import { useState } from "react" +import { useCallback, useState } from "react" import { type AlertsConfig, AlertsService, type Channel } from "@/client" +import { TelegramConnect } from "@/components/Alerts/TelegramConnect" import { InfoTip } from "@/components/Flow/SidePanel" import { Button } from "@/components/ui/button" import { Checkbox } from "@/components/ui/checkbox" @@ -68,6 +69,8 @@ const FIELDS: Record = { // Nothing to configure: it goes to whichever browsers subscribed here, which // is a button rather than a setting. webpush: [], + // The token and the chat are connected by `TelegramConnect`, not typed. + telegram: [], } const KINDS: Channel["kind"][] = [ @@ -76,6 +79,7 @@ const KINDS: Channel["kind"][] = [ "webhook", "dashboard", "webpush", + "telegram", ] /** A setting may hold a `{"$secret": "name"}` reference rather than a literal, @@ -218,6 +222,32 @@ function AlertsForm({ initial }: { initial: AlertsConfig }) { const rules = draft.rules ?? [] const dirty = JSON.stringify(draft) !== JSON.stringify(saved) + // A chat connected on the server: take it into the form without touching + // what has been typed since, and read the rest as stored. + const adopt = useCallback( + (stored: AlertsConfig) => { + setSaved(stored) + setDraft((current) => ({ + ...current, + channels: (current.channels ?? []).map((channel) => { + const fresh = stored.channels?.find((c) => c.name === channel.name) + return fresh?.config?.chat_id && channel.kind === "telegram" + ? { + ...channel, + config: { + ...channel.config, + chat_id: fresh.config.chat_id, + label: fresh.config.label, + }, + } + : channel + }), + })) + queryClient.setQueryData(alertsKey, stored) + }, + [queryClient], + ) + const patchChannel = (index: number, patch: Partial) => setDraft({ ...draft, @@ -359,6 +389,18 @@ function AlertsForm({ initial }: { initial: AlertsConfig }) { {channel.kind === "webpush" ? : null} + {channel.kind === "telegram" ? ( + + patchChannel(index, { + config: { ...channel.config, ...config }, + }) + } + /> + ) : null}
{FIELDS[channel.kind].map(([key, label, placeholder]) => ( diff --git a/frontend/tests/alerts.spec.ts b/frontend/tests/alerts.spec.ts new file mode 100644 index 0000000..8d7fd03 --- /dev/null +++ b/frontend/tests/alerts.spec.ts @@ -0,0 +1,120 @@ +import { expect, test } from "@playwright/test" +import { firstSuperuser, firstSuperuserPassword } from "./config.ts" +import { api, apiPage } from "./utils/api.ts" +import { createUser } from "./utils/privateApi.ts" +import { randomEmail, randomPassword } from "./utils/random.ts" +import { logInUser } from "./utils/user.ts" + +/** + * The Alerts page: channels and rules are a superuser's, and the Telegram + * card asks for a stored token and a saved channel before it connects a chat. + * The bot itself is never reached, so the engine's answers are stubbed. + */ + +test.use({ storageState: { cookies: [], origins: [] } }) +test.describe.configure({ mode: "serial" }) + +const channel = "tg_spec" +let original: { channels?: unknown[] } = {} + +test.beforeAll(async ({ browser }) => { + const page = await apiPage(browser) + original = await (await api(page, "/alerts/config")).json() + await api(page, "/alerts/config", { + method: "PUT", + data: { + ...original, + channels: [ + ...(original.channels ?? []), + { + name: channel, + kind: "telegram", + enabled: true, + config: { token: { $secret: "telegram-bot" } }, + }, + ], + }, + }) + await page.close() +}) + +test.afterAll(async ({ browser }) => { + const page = await apiPage(browser) + await api(page, "/alerts/config", { method: "PUT", data: original }) + await page.close() +}) + +test("a new Telegram channel cannot connect before it has a token", async ({ + page, +}) => { + await logInUser(page, firstSuperuser, firstSuperuserPassword) + await page.goto("/alerts") + + await page.getByTestId("add-channel").click() + await page.getByLabel("Channel kind").last().click() + await page.getByRole("option", { name: "telegram" }).click() + + const card = page.getByTestId("telegram-connect").last() + await expect(card.getByText("Not connected")).toBeVisible() + await expect(card.getByTestId("telegram-connect-chat")).toBeDisabled() + // The token is typed into a password field and kept in the secrets store. + await expect(card.getByLabel("Bot token", { exact: true })).toHaveAttribute( + "type", + "password", + ) +}) + +test("a connected chat is taken into the form without leaving it unsaved", async ({ + page, +}) => { + let linked = false + await page.route("**/api/v1/alerts/telegram/*/link", async (route) => { + if (route.request().method() === "POST") { + return route.fulfill({ + json: { + url: "https://t.me/fluksio_bot?start=code", + group_url: "https://t.me/fluksio_bot?startgroup=code", + expires_in: 600, + }, + }) + } + // The chat answers on the first look. + linked = true + return route.fulfill({ json: { waiting: false, expires_in: 0, error: "" } }) + }) + await page.route("**/api/v1/alerts/config", async (route) => { + const response = await route.fetch() + if (route.request().method() !== "GET" || !linked) { + return route.fulfill({ response }) + } + const stored = await response.json() + for (const c of stored.channels) { + if (c.name === channel) { + c.config = { ...c.config, chat_id: "42", label: "Anna" } + } + } + return route.fulfill({ response, json: stored }) + }) + await logInUser(page, firstSuperuser, firstSuperuserPassword) + await page.goto("/alerts") + + const card = page.getByTestId("telegram-connect").last() + await card.getByTestId("telegram-connect-chat").click() + + await expect(card.getByText("Connected to Anna (private chat)")).toBeVisible() + await expect(page.getByTestId("save-alerts")).toBeDisabled() +}) + +test("an ordinary account sees only the button for its own browser", async ({ + page, +}) => { + const email = randomEmail() + const password = randomPassword() + await createUser({ email, password }) + await logInUser(page, email, password) + + await page.goto("/alerts") + + await expect(page.getByTestId("this-browser")).toBeVisible() + await expect(page.getByTestId("add-channel")).toHaveCount(0) +})