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 <code>` 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.
This commit is contained in:
@@ -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."""
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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<token>/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 <code>``. 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 <code>`` in a private chat, ``/start@bot <code>``
|
||||
#: 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"<b>{html.escape(title)}</b>\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()
|
||||
+11
-4
@@ -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())
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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", "<pump> failed", "a & b <c>", quiet=True)
|
||||
)
|
||||
|
||||
(message,) = bot.sent()
|
||||
assert message["chat_id"] == "42"
|
||||
assert message["parse_mode"] == "HTML"
|
||||
assert message["text"] == "<b><pump> failed</b>\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
|
||||
Reference in New Issue
Block a user