From 5d0ea480bf5313075a78eb03f4e9816b263c9819 Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 28 Oct 2026 13:06:58 +0100 Subject: [PATCH] Alerts: a Telegram channel, one chat or group each, connected by a one-time link The channel sends through the Bot API with httpx only: HTML with escaped text, cut to fit a message, silent for a warning, a short 429 waited out once, and errors that say what to do about a blocked bot, a lost chat or a bad token and never quote the request URL. The bot token is a stored secret: saving a channel refuses a literal one, the card's "Store token" writes the secret and a reference, and no endpoint returns it. A chat id is never typed. Connect starts a ten minute code, and the chat that sends `/start ` back through the bot is the one connected, so it is a chat somebody opened a link for. The engine fetches updates only while a code is waiting (no standing poll, no webhook, no inbound address) and ignores everything else without a reply. Connect needs a saved form, and the form takes the connected chat in without touching what was typed since. The token is part of every Bot API URL, so the httpx logger gets the same filter the request lines have, which now also masks an httpx.URL argument. --- backend/fluksio/api/routes/alerts.py | 116 ++++++- backend/fluksio/flow/alerts.py | 23 +- backend/fluksio/flow/telegram.py | 303 ++++++++++++++++++ backend/fluksio/main.py | 15 +- backend/tests/api/routes/test_alerts.py | 134 ++++++++ backend/tests/flow/test_telegram.py | 276 ++++++++++++++++ docs/code/api.md | 1 + docs/interface/operations.md | 26 ++ frontend/src/client/schemas.gen.ts | 44 ++- frontend/src/client/sdk.gen.ts | 69 +++- frontend/src/client/types.gen.ts | 34 +- .../src/components/Alerts/TelegramConnect.tsx | 268 ++++++++++++++++ frontend/src/routes/_layout/alerts.tsx | 44 ++- frontend/tests/alerts.spec.ts | 120 +++++++ 14 files changed, 1462 insertions(+), 11 deletions(-) create mode 100644 backend/fluksio/flow/telegram.py create mode 100644 backend/tests/flow/test_telegram.py create mode 100644 frontend/src/components/Alerts/TelegramConnect.tsx create mode 100644 frontend/tests/alerts.spec.ts 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) +})