stroblme
5d0ea480bf
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.
304 lines
10 KiB
Python
304 lines
10 KiB
Python
"""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()
|