From 330ee8b8b30cb3955ab6978669eaf839d4dddcac Mon Sep 17 00:00:00 2001 From: "Andrei.Ovcharenko" Date: Sat, 11 Jul 2026 09:07:43 +0300 Subject: [PATCH] fix: harden durable Telegram delivery --- .env.example | 2 + .env.production.example | 2 + .env.systemd.example | 15 + .github/workflows/ci.yml | 7 +- CHANGELOG.md | 6 + README.en.md | 46 +- README.md | 51 +- README.ru.md | 51 +- docs/architecture.md | 21 +- routes.example.json | 4 +- src/telegram_resender/app.py | 227 ++++-- src/telegram_resender/cli.py | 26 +- src/telegram_resender/delivery.py | 12 +- src/telegram_resender/formatting.py | 10 + src/telegram_resender/messages.py | 55 +- src/telegram_resender/models.py | 2 + src/telegram_resender/routes.py | 193 ++++- src/telegram_resender/service.py | 28 +- src/telegram_resender/settings.py | 74 +- src/telegram_resender/storage.py | 857 +++++++++++++++++++++-- src/telegram_resender/telegram_limits.py | 21 + src/telegram_resender/whitelist.py | 26 +- tests/integration/test_bootstrap.py | 20 +- tests/security/test_settings_security.py | 94 ++- tests/unit/test_app.py | 389 +++++++++- tests/unit/test_cli.py | 81 ++- tests/unit/test_delivery.py | 3 + tests/unit/test_formatting.py | 30 + tests/unit/test_routes.py | 169 ++++- tests/unit/test_service.py | 130 ++++ tests/unit/test_storage.py | 358 +++++++++- tests/unit/test_telegram_limits.py | 20 + tests/unit/test_whitelist.py | 8 + 33 files changed, 2812 insertions(+), 226 deletions(-) create mode 100644 .env.systemd.example create mode 100644 src/telegram_resender/telegram_limits.py create mode 100644 tests/unit/test_telegram_limits.py diff --git a/.env.example b/.env.example index 8246480..cb9f1f7 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,8 @@ TELEGRAM_RESENDER_WHITELIST_PATH=whitelist.csv TELEGRAM_RESENDER_ROUTES_PATH= TELEGRAM_RESENDER_LOCALE=ru TELEGRAM_RESENDER_CONFIRM_BEFORE_FORWARD=false +TELEGRAM_RESENDER_PENDING_REQUEST_TTL_SECONDS=900 +TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS=300 TELEGRAM_RESENDER_ADMIN_IDS= TELEGRAM_RESENDER_STORAGE_PATH=telegram_resender.sqlite3 TELEGRAM_RESENDER_DELIVERY_MAX_ATTEMPTS=3 diff --git a/.env.production.example b/.env.production.example index 55e9e8f..e9213a4 100644 --- a/.env.production.example +++ b/.env.production.example @@ -4,6 +4,8 @@ TELEGRAM_RESENDER_WHITELIST_PATH=/data/whitelist.csv TELEGRAM_RESENDER_ROUTES_PATH= TELEGRAM_RESENDER_LOCALE=ru TELEGRAM_RESENDER_CONFIRM_BEFORE_FORWARD=true +TELEGRAM_RESENDER_PENDING_REQUEST_TTL_SECONDS=900 +TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS=300 TELEGRAM_RESENDER_ADMIN_IDS=123456789 TELEGRAM_RESENDER_STORAGE_PATH=/data/telegram_resender.sqlite3 TELEGRAM_RESENDER_DELIVERY_MAX_ATTEMPTS=3 diff --git a/.env.systemd.example b/.env.systemd.example new file mode 100644 index 0000000..f07edc2 --- /dev/null +++ b/.env.systemd.example @@ -0,0 +1,15 @@ +TELEGRAM_RESENDER_BOT_TOKEN=123456:replace-me +TELEGRAM_RESENDER_FORWARD_CHAT_ID=-1001234567890 +TELEGRAM_RESENDER_WHITELIST_PATH=/opt/telegram-resender/data/whitelist.csv +TELEGRAM_RESENDER_ROUTES_PATH= +TELEGRAM_RESENDER_LOCALE=ru +TELEGRAM_RESENDER_CONFIRM_BEFORE_FORWARD=true +TELEGRAM_RESENDER_PENDING_REQUEST_TTL_SECONDS=900 +TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS=300 +TELEGRAM_RESENDER_ADMIN_IDS=123456789 +TELEGRAM_RESENDER_STORAGE_PATH=/opt/telegram-resender/data/telegram_resender.sqlite3 +TELEGRAM_RESENDER_DELIVERY_MAX_ATTEMPTS=3 +TELEGRAM_RESENDER_DELIVERY_RETRY_BACKOFF=1.0 +TELEGRAM_RESENDER_LOG_LEVEL=INFO +TELEGRAM_RESENDER_LOG_FORMAT=JSON +TELEGRAM_RESENDER_POLLING_TIMEOUT=30 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4dd5859..eed1f90 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -113,9 +113,4 @@ jobs: python -m pip install --require-hashes -r requirements-dev.lock python -m pip install --require-hashes -r requirements-audit.lock - name: Pip audit - run: | - # aiogram 3.28.2 pins aiohttp<3.14 while these CVEs are fixed in aiohttp 3.14.0. - # Remove these ignores once aiogram allows aiohttp>=3.14. - python -m pip_audit \ - --ignore-vuln CVE-2026-34993 \ - --ignore-vuln CVE-2026-47265 + run: python -m pip_audit diff --git a/CHANGELOG.md b/CHANGELOG.md index 20c791f..7587dc4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ ## 1.6.2 - Harden storage, message handling, dependency locks, Docker metadata, and CI/release checks after repository audit. +- Add atomic versioned SQLite delivery leases with stale-owner recovery. +- Persist confirmation previews and ownership versions in SQLite across process restarts. +- Add immutable `allowed_user_ids` route filters while retaining username filters for migration. +- Reject duplicate route JSON keys and normalized-empty username filters. +- Add bounded retry guidance after failed confirmed delivery. +- Split Docker and systemd environment templates so their data paths match each runtime. ## 1.6.1 - Apply `ruff format` to match the CI formatter gate after the `1.6.0` release. diff --git a/README.en.md b/README.en.md index d42ae71..dc812be 100644 --- a/README.en.md +++ b/README.en.md @@ -31,7 +31,7 @@ The project is intentionally small and practical: - required-field validation and optional confirmation before forwarding - admin commands for status checks and whitelist reloads without restart - optional multi-route rules through `routes.json` -- SQLite delivery log, retry/backoff, and request-id idempotency +- SQLite delivery/pending state, retry/backoff, and request-id idempotency - Docker, docker-compose, systemd, and health checks for production - deterministic formatting of forwarded messages with request id and submitted time - strict startup validation for secrets and a `doctor` diagnostics command @@ -69,8 +69,10 @@ All settings are read from environment variables (or `.env`). | `TELEGRAM_RESENDER_ROUTES_PATH` | no | JSON route config path | | `TELEGRAM_RESENDER_LOCALE` | no | `ru` or `en`, default `ru` | | `TELEGRAM_RESENDER_CONFIRM_BEFORE_FORWARD` | no | `true` to show a preview and wait for `/confirm` | +| `TELEGRAM_RESENDER_PENDING_REQUEST_TTL_SECONDS` | no | preview lifetime in seconds, default `900` | +| `TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS` | no | stale delivery/confirmation owner timeout, default `300` | | `TELEGRAM_RESENDER_ADMIN_IDS` | no | comma-separated Telegram user IDs allowed to run admin commands | -| `TELEGRAM_RESENDER_STORAGE_PATH` | no | SQLite delivery log path, default `telegram_resender.sqlite3` | +| `TELEGRAM_RESENDER_STORAGE_PATH` | no | SQLite delivery and pending-state path, default `telegram_resender.sqlite3` | | `TELEGRAM_RESENDER_DELIVERY_MAX_ATTEMPTS` | no | Telegram send attempts, default `3` | | `TELEGRAM_RESENDER_DELIVERY_RETRY_BACKOFF` | no | base retry delay in seconds | | `TELEGRAM_RESENDER_REQUEST_ACCEPTED_MESSAGE` | no | text returned on success | @@ -100,10 +102,15 @@ Comment: meeting with facilities ``` The bot currently accepts text only. Media and documents are not forwarded. +If the formatted request or confirmation preview exceeds Telegram's message limit, the bot asks +the user to shorten the comment or other fields. Required fields are building, arrival date/time, vehicle, and license plate. With `TELEGRAM_RESENDER_CONFIRM_BEFORE_FORWARD=true`, the bot shows a preview and only -forwards after `/confirm`. `/cancel` discards the pending request. +forwards after `/confirm`. With multiple previews, target one with +`/confirm ` or `/cancel `; commands without an ID apply to +the latest request. Pending previews are stored in SQLite, survive process restarts, +and expire after the configured TTL. ## Administration @@ -128,7 +135,8 @@ If `TELEGRAM_RESENDER_ROUTES_PATH` is not set, the bot uses a single route from { "name": "tower-a", "target_chat_id": -1002222222222, - "allowed_usernames": ["building_admin"], + "allowed_user_ids": [123456789], + "allowed_usernames": [], "keywords_any": ["Tower A"], "keywords_none": ["cancel"], "template": "[{route}]\n{request}", @@ -139,12 +147,22 @@ If `TELEGRAM_RESENDER_ROUTES_PATH` is not set, the bot uses a single route from ``` The bot forwards a request to every enabled route matching the user and keyword filters. +Use `allowed_user_ids` for authorization-sensitive routing. `allowed_usernames` remains +accepted for migration compatibility, but Telegram usernames are mutable and this legacy +filter must only be used as a non-security label in addition to the numeric-ID whitelist. +If both fields are present, both filters must match. Duplicate JSON keys and malformed or +empty filters are rejected at startup. +When several rules match the same target chat, the first matching rule in the file wins so +that one request produces one deterministic payload per destination. ## Delivery reliability The bot keeps a SQLite delivery log. Each `request_id + target_chat_id` pair stores `pending`, `delivered`, or `failed` status, sender, and the last error. Once a pair is delivered, processing the same request id again will not send a duplicate message. +An atomic SQLite owner/version lease also blocks overlapping handlers from sending the same +pair concurrently. A stale lease can be reclaimed after +`TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS`. ```bash telegram-resender doctor --storage-check @@ -163,8 +181,14 @@ cp whitelist.example.csv data/whitelist.csv docker compose up -d --build ``` -The systemd example is in [deploy/telegram-resender.service](deploy/telegram-resender.service). -The production environment template is [.env.production.example](.env.production.example). +The Docker environment template is [.env.production.example](.env.production.example). +For systemd, use [deploy/telegram-resender.service](deploy/telegram-resender.service) together +with [.env.systemd.example](.env.systemd.example); both use +`/opt/telegram-resender/data` for mutable files. + +```bash +cp .env.systemd.example /opt/telegram-resender/.env.production +``` For production logs, use: @@ -192,6 +216,16 @@ gh attestation verify telegram_resender-*.whl --repo krotname/TelegramResenderBo - This is bot-based intake/forwarding, not a userbot. - The bot does not bypass protected or restricted Telegram chats. - Media, documents, voice messages, and polls are not forwarded as payloads yet. +- Telegram delivery and the following SQLite commit cannot be one atomic operation. A process + crash after Telegram accepts a message but before the delivered state is committed can cause + one retry after the stale lease expires. +- Sending a confirmation preview and publishing its SQLite pending row are likewise not one + transaction. A crash in that narrow interval can leave a visible preview that must be sent + again. Publishing happens after a successful Telegram response so a failed preview never + replaces an older confirmable request. +- Known retry waits extend both ownership leases automatically. An individual Telegram API call + that hangs longer than `TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS` can still be reclaimed; set + the lease above the transport timeout used in the deployment. - AI rewrite, translation, and digest features are not implemented. - Hosted SaaS, mobile apps, and a web dashboard are outside the current self-hosted scope. diff --git a/README.md b/README.md index e843dfd..8bea6ed 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ - Проверка обязательных полей заявки и опциональное подтверждение перед пересылкой. - Админские команды для статуса и перезагрузки whitelist без рестарта. - Опциональные multi-route правила через `routes.json`. -- SQLite delivery log, retry/backoff и идемпотентность по request id. +- SQLite delivery/pending state, retry/backoff и идемпотентность по request id. - Docker, docker-compose, systemd и health-check для production. - Детерминированный формат пересылки с request id и временем заявки. @@ -60,8 +60,10 @@ telegram-resender | `TELEGRAM_RESENDER_ROUTES_PATH` | нет | путь к JSON-файлу маршрутов | | `TELEGRAM_RESENDER_LOCALE` | нет | `ru` или `en`, по умолчанию `ru` | | `TELEGRAM_RESENDER_CONFIRM_BEFORE_FORWARD` | нет | `true`, чтобы показывать preview и ждать `/confirm` | +| `TELEGRAM_RESENDER_PENDING_REQUEST_TTL_SECONDS` | нет | срок жизни preview в секундах, по умолчанию `900` | +| `TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS` | нет | timeout stale-владельца доставки/подтверждения, по умолчанию `300` | | `TELEGRAM_RESENDER_ADMIN_IDS` | нет | Telegram user id администраторов через запятую | -| `TELEGRAM_RESENDER_STORAGE_PATH` | нет | SQLite delivery log, по умолчанию `telegram_resender.sqlite3` | +| `TELEGRAM_RESENDER_STORAGE_PATH` | нет | SQLite delivery/pending state, по умолчанию `telegram_resender.sqlite3` | | `TELEGRAM_RESENDER_DELIVERY_MAX_ATTEMPTS` | нет | попытки отправки в Telegram, по умолчанию `3` | | `TELEGRAM_RESENDER_DELIVERY_RETRY_BACKOFF` | нет | базовая задержка retry в секундах | | `TELEGRAM_RESENDER_LOG_LEVEL` | нет | `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL` | @@ -73,8 +75,8 @@ telegram-resender Формат whitelist (по одному неизменяемому числовому Telegram user ID в строке; ID можно узнать через `/whoami`): ```csv -alice -@bob +123456789 +987654321 ``` ## Формат заявки @@ -90,11 +92,15 @@ alice ``` Пока бот принимает только текст. Медиа и документы не пересылаются. +Если заявка со служебным заголовком или preview превышает лимит Telegram, бот попросит +сократить комментарий или другие поля. Обязательные поля: объект/здание, дата и время прибытия, автомобиль, госномер. Если включить `TELEGRAM_RESENDER_CONFIRM_BEFORE_FORWARD=true`, бот покажет preview -заявки и отправит ее администратору только после `/confirm`. Команда `/cancel` -отменяет ожидающую подтверждения заявку. +заявки и отправит ее администратору только после `/confirm`. Для нескольких preview +можно указать ID: `/confirm ` или `/cancel `; команды без ID +относятся к последней заявке. Pending preview хранится в SQLite, переживает рестарт +процесса и истекает через настроенный TTL. ## Администрирование @@ -119,7 +125,8 @@ alice { "name": "tower-a", "target_chat_id": -1002222222222, - "allowed_usernames": ["building_admin"], + "allowed_user_ids": [123456789], + "allowed_usernames": [], "keywords_any": ["Башня А"], "keywords_none": ["отмена"], "template": "[{route}]\n{request}", @@ -130,6 +137,13 @@ alice ``` Заявка отправляется во все активные маршруты, где совпали пользователь и keyword-фильтры. +Для маршрутизации с контролем доступа используйте `allowed_user_ids`. Поле +`allowed_usernames` сохранено для миграционной совместимости, но Telegram username изменяем: +этот legacy-фильтр допустим только как незащитная метка поверх общего числового whitelist. +Если заданы оба поля, должны совпасть оба. Дубли JSON-ключей, пустые и некорректные +фильтры отклоняются при старте. +Если одному target chat соответствуют несколько правил, применяется первое совпавшее +правило из файла: одна заявка формирует один детерминированный payload на destination. ## Надежность доставки @@ -137,6 +151,9 @@ alice хранится статус `pending`, `delivered` или `failed`, отправитель и последняя ошибка. Если такая пара уже доставлена, повторная обработка того же request id не отправит сообщение второй раз. +Атомарный SQLite lease с owner/version также не дает параллельным handler-задачам +одновременно отправить одну пару. Stale lease можно занять заново после +`TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS`. ```bash telegram-resender doctor --storage-check @@ -155,8 +172,14 @@ copy whitelist.example.csv data\whitelist.csv docker compose up -d --build ``` -Systemd пример лежит в [deploy/telegram-resender.service](deploy/telegram-resender.service). -Production env template: [.env.production.example](.env.production.example). +Docker env template: [.env.production.example](.env.production.example). Для systemd +используйте [deploy/telegram-resender.service](deploy/telegram-resender.service) вместе с +[.env.systemd.example](.env.systemd.example): оба используют +`/opt/telegram-resender/data` для изменяемых файлов. + +```bash +cp .env.systemd.example /opt/telegram-resender/.env.production +``` Для production-логов включите: @@ -184,6 +207,16 @@ gh attestation verify telegram_resender-*.whl --repo krotname/TelegramResenderBo - Это bot-based intake/forwarding, не userbot. - Бот не обходит protected/restricted Telegram-чаты. - Медиа, документы, voice и polls пока не пересылаются как payload. +- Отправку в Telegram и следующий SQLite commit невозможно сделать одной атомарной операцией. + Если процесс завершится после приема сообщения Telegram, но до записи `delivered`, возможна + одна повторная отправка после истечения stale lease. +- Отправка confirmation preview и публикация pending-строки в SQLite тоже не являются одной + транзакцией. Сбой в узком промежутке может оставить видимый preview, который придется отправить + заново. Pending публикуется после успешного ответа Telegram, поэтому неотправленный preview не + заменяет более старую подтверждаемую заявку. +- Известные retry waits автоматически продлевают оба lease. Но отдельный Telegram API call, + зависший дольше `TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS`, может быть занят заново; задавайте + lease больше transport timeout текущего deployment. - AI rewrite/translate/digest не реализованы. - Hosted SaaS, mobile app и web dashboard не входят в текущую self-hosted область. diff --git a/README.ru.md b/README.ru.md index 4a7dcf5..7dfb9da 100644 --- a/README.ru.md +++ b/README.ru.md @@ -29,7 +29,7 @@ - Проверка обязательных полей заявки и опциональное подтверждение перед пересылкой. - Админские команды для статуса и перезагрузки whitelist без рестарта. - Опциональные multi-route правила через `routes.json`. -- SQLite delivery log, retry/backoff и идемпотентность по request id. +- SQLite delivery/pending state, retry/backoff и идемпотентность по request id. - Docker, docker-compose, systemd и health-check для production. - Детерминированный формат пересылки с request id и временем заявки. @@ -57,8 +57,10 @@ telegram-resender | `TELEGRAM_RESENDER_ROUTES_PATH` | нет | путь к JSON-файлу маршрутов | | `TELEGRAM_RESENDER_LOCALE` | нет | `ru` или `en`, по умолчанию `ru` | | `TELEGRAM_RESENDER_CONFIRM_BEFORE_FORWARD` | нет | `true`, чтобы показывать preview и ждать `/confirm` | +| `TELEGRAM_RESENDER_PENDING_REQUEST_TTL_SECONDS` | нет | срок жизни preview в секундах, по умолчанию `900` | +| `TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS` | нет | timeout stale-владельца доставки/подтверждения, по умолчанию `300` | | `TELEGRAM_RESENDER_ADMIN_IDS` | нет | Telegram user id администраторов через запятую | -| `TELEGRAM_RESENDER_STORAGE_PATH` | нет | SQLite delivery log, по умолчанию `telegram_resender.sqlite3` | +| `TELEGRAM_RESENDER_STORAGE_PATH` | нет | SQLite delivery/pending state, по умолчанию `telegram_resender.sqlite3` | | `TELEGRAM_RESENDER_DELIVERY_MAX_ATTEMPTS` | нет | попытки отправки в Telegram, по умолчанию `3` | | `TELEGRAM_RESENDER_DELIVERY_RETRY_BACKOFF` | нет | базовая задержка retry в секундах | | `TELEGRAM_RESENDER_LOG_LEVEL` | нет | `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL` | @@ -70,8 +72,8 @@ telegram-resender Формат whitelist (по одному неизменяемому числовому Telegram user ID в строке; ID можно узнать через `/whoami`): ```csv -alice -@bob +123456789 +987654321 ``` ## Формат заявки @@ -87,11 +89,15 @@ alice ``` Пока бот принимает только текст. Медиа и документы не пересылаются. +Если заявка со служебным заголовком или preview превышает лимит Telegram, бот попросит +сократить комментарий или другие поля. Обязательные поля: объект/здание, дата и время прибытия, автомобиль, госномер. Если включить `TELEGRAM_RESENDER_CONFIRM_BEFORE_FORWARD=true`, бот покажет preview -заявки и отправит ее администратору только после `/confirm`. Команда `/cancel` -отменяет ожидающую подтверждения заявку. +заявки и отправит ее администратору только после `/confirm`. Для нескольких preview +можно указать ID: `/confirm ` или `/cancel `; команды без ID +относятся к последней заявке. Pending preview хранится в SQLite, переживает рестарт +процесса и истекает через настроенный TTL. ## Администрирование @@ -116,7 +122,8 @@ alice { "name": "tower-a", "target_chat_id": -1002222222222, - "allowed_usernames": ["building_admin"], + "allowed_user_ids": [123456789], + "allowed_usernames": [], "keywords_any": ["Башня А"], "keywords_none": ["отмена"], "template": "[{route}]\n{request}", @@ -127,6 +134,13 @@ alice ``` Заявка отправляется во все активные маршруты, где совпали пользователь и keyword-фильтры. +Для маршрутизации с контролем доступа используйте `allowed_user_ids`. Поле +`allowed_usernames` сохранено для миграционной совместимости, но Telegram username изменяем: +этот legacy-фильтр допустим только как незащитная метка поверх общего числового whitelist. +Если заданы оба поля, должны совпасть оба. Дубли JSON-ключей, пустые и некорректные +фильтры отклоняются при старте. +Если одному target chat соответствуют несколько правил, применяется первое совпавшее +правило из файла: одна заявка формирует один детерминированный payload на destination. ## Надежность доставки @@ -134,6 +148,9 @@ alice хранится статус `pending`, `delivered` или `failed`, отправитель и последняя ошибка. Если такая пара уже доставлена, повторная обработка того же request id не отправит сообщение второй раз. +Атомарный SQLite lease с owner/version также не дает параллельным handler-задачам +одновременно отправить одну пару. Stale lease можно занять заново после +`TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS`. ```bash telegram-resender doctor --storage-check @@ -152,8 +169,14 @@ copy whitelist.example.csv data\whitelist.csv docker compose up -d --build ``` -Systemd пример лежит в [deploy/telegram-resender.service](deploy/telegram-resender.service). -Production env template: [.env.production.example](.env.production.example). +Docker env template: [.env.production.example](.env.production.example). Для systemd +используйте [deploy/telegram-resender.service](deploy/telegram-resender.service) вместе с +[.env.systemd.example](.env.systemd.example): оба используют +`/opt/telegram-resender/data` для изменяемых файлов. + +```bash +cp .env.systemd.example /opt/telegram-resender/.env.production +``` Для production-логов включите: @@ -181,6 +204,16 @@ gh attestation verify telegram_resender-*.whl --repo krotname/TelegramResenderBo - Это bot-based intake/forwarding, не userbot. - Бот не обходит protected/restricted Telegram-чаты. - Медиа, документы, voice и polls пока не пересылаются как payload. +- Отправку в Telegram и следующий SQLite commit невозможно сделать одной атомарной операцией. + Если процесс завершится после приема сообщения Telegram, но до записи `delivered`, возможна + одна повторная отправка после истечения stale lease. +- Отправка confirmation preview и публикация pending-строки в SQLite тоже не являются одной + транзакцией. Сбой в узком промежутке может оставить видимый preview, который придется отправить + заново. Pending публикуется после успешного ответа Telegram, поэтому неотправленный preview не + заменяет более старую подтверждаемую заявку. +- Известные retry waits автоматически продлевают оба lease. Но отдельный Telegram API call, + зависший дольше `TELEGRAM_RESENDER_DELIVERY_LEASE_SECONDS`, может быть занят заново; задавайте + lease больше transport timeout текущего deployment. - AI rewrite/translate/digest не реализованы. - Hosted SaaS, mobile app и web dashboard не входят в текущую self-hosted область. diff --git a/docs/architecture.md b/docs/architecture.md index 208e5ae..706ef24 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -3,22 +3,27 @@ ```text User Telegram Message -> app.incoming_from_message -> service.ResenderService.handle_text -> [whitelist.check + template parsing + required-field validation] -> - - missing username: bot replies with chat-id guidance - - unknown username: bot replies with private-bot denial + - missing sender ID: bot replies with chat-id guidance + - unknown user ID: bot replies with private-bot denial - incomplete request: bot replies with missing-field guidance - no route matched: bot replies without forwarding - allowed: formatter.MessageFormatter.format_forward -> delivery log -> matching target chats - - confirmation mode: preview -> /confirm -> target chat, or /cancel + - confirmation mode: successful preview -> SQLite pending state with TTL/version ownership -> + /confirm [request-id] -> atomic claim -> current-whitelist recheck -> target chat, + or /cancel [request-id]; pending state survives process restarts ``` - `settings.Settings`: validates runtime environment and normalizes paths. - `whitelist.Whitelist`: file-backed access control. - `service.ResenderService`: pure decision function. - `requests.py`: request-template parsing and required-field validation. -- `routes.py`: optional JSON route loading and route match predicates. +- `routes.py`: strict duplicate-safe JSON route loading, numeric-ID authorization filters, + and legacy non-security username route labels. - `formatting.MessageFormatter`: stable forwarding format. - `delivery.py`: retry/backoff wrapper for Telegram send failures. -- `storage.py`: SQLite delivery log and CSV export. +- `storage.py`: SQLite delivery log, owner/version leases, persistent pending previews, + schema migration/validation, and CSV export. +- `telegram_limits.py`: shared Telegram UTF-16 message-length rules. - `app.py`: adapter layer to aiogram (`Message` -> domain model). - `messages.py`: localized user-facing message catalogs. - `cli.py`: explicit application entrypoint, `doctor`, `health`, and CSV export commands. @@ -38,8 +43,10 @@ Delivery flow: ```text matched route target -> storage.RequestLog.begin_delivery -> - already delivered: skip duplicate send - pending: delivery.send_with_retry -> mark delivered or failed + already delivered or actively leased: skip duplicate send + atomic owner/version lease: delivery.send_with_retry -> owner-checked delivered/failed state + stale lease: a newer version may reclaim ownership after the configured timeout +health/doctor -> verify delivery, lease, pending, and sequence table schemas ``` Deployment surface: diff --git a/routes.example.json b/routes.example.json index 509a113..9851cb3 100644 --- a/routes.example.json +++ b/routes.example.json @@ -3,6 +3,7 @@ { "name": "default", "target_chat_id": -1001234567890, + "allowed_user_ids": [], "allowed_usernames": [], "keywords_any": [], "keywords_none": [], @@ -11,7 +12,8 @@ { "name": "tower-a", "target_chat_id": -1002222222222, - "allowed_usernames": ["building_admin"], + "allowed_user_ids": [123456789], + "allowed_usernames": [], "keywords_any": ["Башня А", "Tower A"], "keywords_none": ["отмена", "cancel"], "template": "[{route}]\n{request}", diff --git a/src/telegram_resender/app.py b/src/telegram_resender/app.py index dc0d145..ac91e05 100644 --- a/src/telegram_resender/app.py +++ b/src/telegram_resender/app.py @@ -4,7 +4,7 @@ import json import logging -from dataclasses import dataclass +from collections.abc import Callable from datetime import UTC, datetime from aiogram import Bot, Dispatcher, F, Router @@ -18,7 +18,14 @@ from telegram_resender.routes import default_route, load_routes from telegram_resender.service import ResenderService from telegram_resender.settings import Settings -from telegram_resender.storage import RequestLog +from telegram_resender.storage import ( + DeliveryInProgressError, + DeliveryLease, + PendingRequestKey, + PendingRequestStore, + RequestLog, +) +from telegram_resender.telegram_limits import fits_telegram_message from telegram_resender.whitelist import Whitelist LOGGER = logging.getLogger(__name__) @@ -39,15 +46,6 @@ def format(self, record: logging.LogRecord) -> str: return json.dumps(payload, ensure_ascii=False) -@dataclass(frozen=True, slots=True) -class PendingRequest: - """Forwarding payload waiting for explicit user confirmation.""" - - request_id: str - forward_text_by_chat_id: tuple[tuple[int, str], ...] - sender_username: str | None - - def incoming_from_message(message: Message) -> IncomingMessage: """Convert an aiogram message into the SDK-free domain model.""" @@ -95,8 +93,15 @@ def create_router(settings: Settings, service: ResenderService) -> Router: router = Router(name="telegram-resender") messages = settings.messages - request_log = RequestLog(settings.storage_path) - pending_requests: dict[int, PendingRequest] = {} + request_log = RequestLog( + settings.storage_path, + lease_seconds=settings.delivery_lease_seconds, + ) + pending_requests = PendingRequestStore( + settings.storage_path, + ttl_seconds=settings.pending_request_ttl_seconds, + claim_seconds=settings.delivery_lease_seconds, + ) @router.message(Command("start")) async def start(message: Message) -> None: @@ -162,28 +167,81 @@ async def reload_whitelist(message: Message) -> None: @router.message(Command("confirm")) async def confirm_request(message: Message) -> None: - pending = pending_requests.pop(message.chat.id, None) + pending_key = _pending_request_key(message) + pending = pending_requests.claim( + pending_key, + request_id=_command_request_id(message, command="confirm"), + ) if pending is None: await message.answer(messages.no_pending_request) return - bot = message.bot - if bot is None: - msg = "Telegram message is not bound to a bot" - raise RuntimeError(msg) - await _deliver_payloads( - bot=bot, - request_log=request_log, - request_id=pending.request_id, - forward_text_by_chat_id=pending.forward_text_by_chat_id, - sender_username=pending.sender_username, - settings=settings, - ) + if not service.is_authorized(pending.sender_user_id): + pending_requests.discard_all(pending_key) + await message.answer(_access_denied_text(message, settings)) + return + try: + bot = message.bot + if bot is None: + msg = "Telegram message is not bound to a bot" + raise RuntimeError(msg) + await _deliver_payloads( + bot=bot, + request_log=request_log, + request_id=pending.request_id, + request_id_aliases=pending.request_id_aliases, + forward_text_by_chat_id=pending.forward_text_by_chat_id, + sender_username=pending.sender_username, + settings=settings, + pending_keepalive=lambda wait_seconds: pending_requests.renew( + pending_key, + pending, + wait_seconds=wait_seconds, + ), + ) + except DeliveryInProgressError: + restored = pending_requests.release(pending_key, pending) + await message.answer( + messages.request_in_progress + if restored + else messages.request_confirmation_failed_expired.format( + request_id=pending.request_id + ) + ) + return + except Exception: + LOGGER.exception("Confirmed request delivery failed") + restored = False + if service.is_authorized(pending.sender_user_id): + try: + restored = pending_requests.release(pending_key, pending) + except Exception: + LOGGER.exception("Failed to release pending request after delivery error") + guidance = ( + messages.request_confirmation_failed_retry.format(request_id=pending.request_id) + if restored + else messages.request_confirmation_failed_expired.format( + request_id=pending.request_id + ) + ) + try: + await message.answer(guidance) + except Exception: + LOGGER.exception("Failed to send confirmation retry guidance") + raise + if not pending_requests.complete(pending_key, pending): + LOGGER.warning( + "Pending request ownership changed before completion: %s", + pending.request_id, + ) LOGGER.info("Forwarded confirmed request from Telegram user") await message.answer(messages.request_confirmed.format(request_id=pending.request_id)) @router.message(Command("cancel")) async def cancel_request(message: Message) -> None: - pending = pending_requests.pop(message.chat.id, None) + pending = pending_requests.cancel( + _pending_request_key(message), + request_id=_command_request_id(message, command="cancel"), + ) if pending is None: await message.answer(messages.no_pending_request) return @@ -194,28 +252,36 @@ async def forward_text(message: Message) -> None: incoming = incoming_from_message(message) decision = service.handle_text(incoming) if decision.should_forward and decision.forward_text is not None: - forward_payloads = tuple( + forward_payloads = decision.forward_payloads or tuple( ( chat_id, service.render_forward_text_for_target(chat_id, decision.forward_text), ) for chat_id in decision.target_chat_ids ) + if any(not fits_telegram_message(text) for _, text in forward_payloads): + await message.answer(messages.request_too_long) + return if settings.confirm_before_forward: if decision.request_id is None: msg = "Forwarding decision is missing request_id" raise RuntimeError(msg) - pending_requests[message.chat.id] = PendingRequest( + confirmation_text = messages.confirmation_prompt.format( + request_id=decision.request_id, + preview=decision.forward_text, + ) + if not fits_telegram_message(confirmation_text): + await message.answer(messages.request_too_long) + return + await message.answer(confirmation_text) + pending_requests.publish( + _pending_request_key(message), request_id=decision.request_id, + request_id_aliases=decision.request_id_aliases, forward_text_by_chat_id=forward_payloads, + sender_user_id=incoming.user.id, sender_username=incoming.user.username, ) - await message.answer( - messages.confirmation_prompt.format( - request_id=decision.request_id, - preview=decision.forward_text, - ) - ) return bot = message.bot if bot is None: @@ -224,14 +290,19 @@ async def forward_text(message: Message) -> None: if decision.request_id is None: msg = "Forwarding decision is missing request_id" raise RuntimeError(msg) - await _deliver_payloads( - bot=bot, - request_log=request_log, - request_id=decision.request_id, - forward_text_by_chat_id=forward_payloads, - sender_username=incoming.user.username, - settings=settings, - ) + try: + await _deliver_payloads( + bot=bot, + request_log=request_log, + request_id=decision.request_id, + request_id_aliases=decision.request_id_aliases, + forward_text_by_chat_id=forward_payloads, + sender_username=incoming.user.username, + settings=settings, + ) + except DeliveryInProgressError: + await message.answer(messages.request_in_progress) + return LOGGER.info("Forwarded message from Telegram user") else: LOGGER.info("Rejected message: %s", decision.reason) @@ -249,6 +320,31 @@ def _telegram_user_id(message: Message) -> int | None: return user.id if user else None +def _pending_request_key(message: Message) -> PendingRequestKey: + """Keep confirmations isolated per user, including inside shared chats.""" + + return message.chat.id, _telegram_user_id(message) + + +def _command_request_id(message: Message, *, command: str) -> str | None: + text = (message.text or "").strip() + if not text: + return None + command_parts = text.split(maxsplit=1) + command_token = command_parts[0] + bare_command = command_token.partition("@")[0].casefold() + if bare_command != f"/{command}" or len(command_parts) == 1: + return None + return command_parts[1].strip() or None + + +def _access_denied_text(message: Message, settings: Settings) -> str: + user_id = _telegram_user_id(message) + if user_id is None: + return settings.messages.access_denied_missing_username.format(chat_id=message.chat.id) + return settings.messages.access_denied_unknown + + def _is_admin(message: Message, admin_ids: frozenset[int]) -> bool: user_id = _telegram_user_id(message) return user_id in admin_ids if user_id is not None else False @@ -259,40 +355,59 @@ async def _deliver_payloads( bot: Bot, request_log: RequestLog, request_id: str, + request_id_aliases: tuple[str, ...], forward_text_by_chat_id: tuple[tuple[int, str], ...], sender_username: str | None, settings: Settings, + pending_keepalive: Callable[[float], bool] | None = None, ) -> None: for chat_id, forward_text in forward_text_by_chat_id: - should_send = request_log.begin_delivery( + lease = request_log.begin_delivery( request_id=request_id, target_chat_id=chat_id, sender_username=sender_username, + request_id_aliases=request_id_aliases, ) - if not should_send: + if lease is None: LOGGER.info("Skipped already delivered request %s to %s", request_id, chat_id) continue + + def renew_ownership( + wait_seconds: float, + owned_lease: DeliveryLease = lease, + ) -> None: + if not request_log.renew_delivery(owned_lease, wait_seconds=wait_seconds): + msg = "Telegram delivery ownership was superseded before retry" + raise RuntimeError(msg) + if pending_keepalive is not None and not pending_keepalive(wait_seconds): + msg = "Pending confirmation ownership was superseded before retry" + raise RuntimeError(msg) + try: + renew_ownership(0.0) await send_with_retry( bot.send_message, chat_id=chat_id, text=forward_text, max_attempts=settings.delivery_max_attempts, backoff_seconds=settings.delivery_retry_backoff, + before_retry_wait=renew_ownership, ) except Exception as exc: - request_log.mark_delivery( - request_id=request_id, - target_chat_id=chat_id, - status="failed", - error=str(exc), - ) + try: + marked = request_log.mark_delivery( + lease=lease, + status="failed", + error=str(exc), + ) + if not marked: + LOGGER.warning("Delivery failure lease was already superseded") + except Exception: + LOGGER.exception("Failed to persist Telegram delivery failure") raise - request_log.mark_delivery( - request_id=request_id, - target_chat_id=chat_id, - status="delivered", - ) + if not request_log.mark_delivery(lease=lease, status="delivered"): + msg = "Telegram delivery completed after its ownership lease was superseded" + raise RuntimeError(msg) def create_dispatcher(settings: Settings, service: ResenderService) -> Dispatcher: diff --git a/src/telegram_resender/cli.py b/src/telegram_resender/cli.py index a87a852..cf75d87 100644 --- a/src/telegram_resender/cli.py +++ b/src/telegram_resender/cli.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse +import sqlite3 import sys from asyncio import run as asyncio_run from collections.abc import Sequence @@ -39,7 +40,7 @@ def run_bot(*, debug: bool = False) -> int: try: asyncio_run(run_polling()) - except (ValidationError, FileNotFoundError, TelegramAPIError, ValueError) as exc: + except (ValidationError, OSError, sqlite3.Error, TelegramAPIError, ValueError) as exc: _print_startup_error(exc, debug=debug) return 2 return 0 @@ -62,11 +63,14 @@ def run_doctor( else (default_route(settings.forward_chat_id),) ) if storage_check: - RequestLog(settings.storage_path).check() + RequestLog( + settings.storage_path, + lease_seconds=settings.delivery_lease_seconds, + ).check() except ValidationError as exc: print(_format_validation_error(exc), file=stderr) return 2 - except (FileNotFoundError, ValueError) as exc: + except (OSError, sqlite3.Error, ValueError) as exc: print(f"Configuration error: {exc}", file=stderr) print(_setup_hint(), file=stderr) return 2 @@ -83,6 +87,8 @@ def run_doctor( print(f"Storage check: {'OK' if storage_check else 'not requested'}", file=stdout) print(f"Admin users: {len(settings.admin_ids)}", file=stdout) print(f"Confirm before forward: {settings.confirm_before_forward}", file=stdout) + print(f"Pending request TTL: {settings.pending_request_ttl_seconds}s", file=stdout) + print(f"Delivery lease: {settings.delivery_lease_seconds}s", file=stdout) print(f"Polling timeout: {settings.polling_timeout}s", file=stdout) return 0 @@ -119,8 +125,11 @@ def run_health(stdout: TextIO = sys.stdout, stderr: TextIO = sys.stderr) -> int: Whitelist.from_file(settings.whitelist_path) if settings.routes_path is not None: load_routes(settings.routes_path) - RequestLog(settings.storage_path).check() - except (ValidationError, FileNotFoundError, ValueError, OSError) as exc: + RequestLog( + settings.storage_path, + lease_seconds=settings.delivery_lease_seconds, + ).check() + except (ValidationError, OSError, sqlite3.Error, ValueError) as exc: print(f"health=error error={exc}", file=stderr) return 2 print("health=ok", file=stdout) @@ -138,8 +147,11 @@ def export_requests( try: settings = Settings() # type: ignore[call-arg] since_date = date.fromisoformat(since) - records = RequestLog(settings.storage_path).records_since(since_date) - except (ValidationError, ValueError, OSError) as exc: + records = RequestLog( + settings.storage_path, + lease_seconds=settings.delivery_lease_seconds, + ).records_since(since_date) + except (ValidationError, OSError, sqlite3.Error, ValueError) as exc: print(f"Export error: {exc}", file=stderr) return 2 export_records_csv(records, stdout) diff --git a/src/telegram_resender/delivery.py b/src/telegram_resender/delivery.py index cdd7c57..3a066a0 100644 --- a/src/telegram_resender/delivery.py +++ b/src/telegram_resender/delivery.py @@ -9,6 +9,7 @@ SendMessage = Callable[[int, str], Awaitable[object]] Sleep = Callable[[float], Awaitable[None]] +BeforeRetryWait = Callable[[float], None] async def send_with_retry( @@ -19,6 +20,7 @@ async def send_with_retry( max_attempts: int, backoff_seconds: float, sleep: Sleep = asyncio.sleep, + before_retry_wait: BeforeRetryWait | None = None, ) -> None: """Send a Telegram message with bounded retry/backoff for API failures.""" @@ -30,9 +32,15 @@ async def send_with_retry( except TelegramRetryAfter as exc: if attempt >= max_attempts: raise - await sleep(float(exc.retry_after)) + delay = float(exc.retry_after) + if before_retry_wait is not None: + before_retry_wait(delay) + await sleep(delay) except TelegramAPIError: if attempt >= max_attempts: raise - await sleep(backoff_seconds * attempt) + delay = backoff_seconds * attempt + if before_retry_wait is not None: + before_retry_wait(delay) + await sleep(delay) attempt += 1 diff --git a/src/telegram_resender/formatting.py b/src/telegram_resender/formatting.py index eb61e07..0b2e80a 100644 --- a/src/telegram_resender/formatting.py +++ b/src/telegram_resender/formatting.py @@ -35,6 +35,16 @@ def format_request_id(self, message: IncomingMessage) -> str: if message.message_id is not None: return f"tg-{message.chat_id}-{message.message_id}" + fingerprint = hashlib.sha256( + f"{message.chat_id}|{message.user.id!r}|{message.text}".encode() + ).hexdigest() + return f"local-{fingerprint[:12]}" + + def format_legacy_request_id(self, message: IncomingMessage) -> str | None: + """Return the legacy fallback id so persisted deliveries remain idempotent.""" + + if message.message_id is not None: + return None fingerprint = hashlib.sha256( f"{message.chat_id}|{message.user.username or ''}|{message.text}".encode() ).hexdigest() diff --git a/src/telegram_resender/messages.py b/src/telegram_resender/messages.py index 56a6fd8..c7868f4 100644 --- a/src/telegram_resender/messages.py +++ b/src/telegram_resender/messages.py @@ -33,11 +33,15 @@ class MessageCatalog: access_denied_missing_username: str unsupported_message: str invalid_request: str + request_too_long: str + request_in_progress: str missing_fields: str no_route_matched: str confirmation_prompt: str request_confirmed: str request_cancelled: str + request_confirmation_failed_retry: str + request_confirmation_failed_expired: str no_pending_request: str admin_access_denied: str admin_status: str @@ -73,11 +77,12 @@ def with_overrides( template=REQUEST_TEMPLATE_RU, request_accepted="Заявка принята и передана администратору.", access_denied_unknown=( - "Этот бот закрытый. Попросите администратора добавить ваш Telegram username в белый список." + "Этот бот закрытый. Попросите администратора добавить ваш числовой Telegram user ID " + "в белый список. Узнать ID можно командой /whoami." ), access_denied_missing_username=( - "У вас не задан Telegram username. Создайте username в настройках Telegram " - "или попросите администратора включить доступ по chat id: {chat_id}." + "Не удалось определить Telegram user ID отправителя. Обратитесь к администратору " + "и сообщите для диагностики chat id: {chat_id}." ), unsupported_message=( "Пока я принимаю только текстовые заявки. Отправьте текст по шаблону из /template." @@ -86,6 +91,14 @@ def with_overrides( "Заявка выглядит неполной. Отправьте объект, время прибытия, модель автомобиля " "и госномер. Шаблон доступен по команде /template." ), + request_too_long=( + "Заявка слишком длинная для пересылки в Telegram. Сократите комментарий или другие " + "поля и отправьте заявку повторно." + ), + request_in_progress=( + "Эта заявка уже передается администратору. Дождитесь результата и не отправляйте " + "ее повторно." + ), missing_fields=( "Заявка неполная. Заполните обязательные поля: {fields}. " "Шаблон доступен по команде /template." @@ -93,10 +106,19 @@ def with_overrides( no_route_matched="Заявка заполнена, но для нее не найден активный маршрут пересылки.", confirmation_prompt=( "Проверьте заявку {request_id}:\n\n{preview}\n\n" - "Отправьте /confirm, чтобы передать ее администратору, или /cancel, чтобы отменить." + "Отправьте /confirm {request_id}, чтобы передать ее администратору, " + "или /cancel {request_id}, чтобы отменить. Команды без ID применяются к последней заявке." ), request_confirmed="Заявка {request_id} подтверждена и передана администратору.", request_cancelled="Заявка {request_id} отменена.", + request_confirmation_failed_retry=( + "Не удалось передать заявку {request_id}. Она сохранена: повторите " + "/confirm {request_id} позже." + ), + request_confirmation_failed_expired=( + "Не удалось передать заявку {request_id}, а срок ее подтверждения уже истек. " + "Отправьте заявку заново." + ), no_pending_request="Нет заявки, ожидающей подтверждения.", admin_access_denied="Эта команда доступна только администратору.", admin_status=( @@ -126,11 +148,12 @@ def with_overrides( template=REQUEST_TEMPLATE_EN, request_accepted="Your request has been accepted and sent to the administrator.", access_denied_unknown=( - "This bot is private. Ask an administrator to add your Telegram username." + "This bot is private. Ask an administrator to add your numeric Telegram user ID " + "to the whitelist. Use /whoami to discover the ID." ), access_denied_missing_username=( - "Your Telegram username is not set. Create one in Telegram settings or ask " - "the administrator to allow your chat id: {chat_id}." + "The sender's Telegram user ID could not be determined. Contact an administrator " + "and provide this chat ID for diagnostics: {chat_id}." ), unsupported_message=( "I can only accept text requests for now. Send a text message using /template." @@ -139,6 +162,13 @@ def with_overrides( "The request looks incomplete. Send the building, arrival time, vehicle model, " "and license plate. Use /template for the expected format." ), + request_too_long=( + "The request is too long to forward through Telegram. Shorten the comment or other " + "fields and send it again." + ), + request_in_progress=( + "This request is already being delivered. Wait for the result instead of sending it again." + ), missing_fields=( "The request is incomplete. Fill these required fields: {fields}. " "Use /template for the expected format." @@ -146,10 +176,19 @@ def with_overrides( no_route_matched="The request is complete, but no active forwarding route matched it.", confirmation_prompt=( "Review request {request_id}:\n\n{preview}\n\n" - "Send /confirm to forward it to the administrator or /cancel to discard it." + "Send /confirm {request_id} to forward it to the administrator or " + "/cancel {request_id} to discard it. Commands without an ID apply to the latest request." ), request_confirmed="Request {request_id} has been confirmed and sent to the administrator.", request_cancelled="Request {request_id} has been cancelled.", + request_confirmation_failed_retry=( + "Request {request_id} could not be delivered but remains saved. " + "Retry /confirm {request_id} later." + ), + request_confirmation_failed_expired=( + "Request {request_id} could not be delivered and its confirmation window has expired. " + "Send the request again." + ), no_pending_request="There is no request waiting for confirmation.", admin_access_denied="This command is available to administrators only.", admin_status=( diff --git a/src/telegram_resender/models.py b/src/telegram_resender/models.py index 6402f5b..c9999d5 100644 --- a/src/telegram_resender/models.py +++ b/src/telegram_resender/models.py @@ -47,3 +47,5 @@ class ForwardingDecision: forward_text: str | None = None request_id: str | None = None target_chat_ids: tuple[int, ...] = () + forward_payloads: tuple[tuple[int, str], ...] = () + request_id_aliases: tuple[str, ...] = () diff --git a/src/telegram_resender/routes.py b/src/telegram_resender/routes.py index 6c0800c..d69bf1f 100644 --- a/src/telegram_resender/routes.py +++ b/src/telegram_resender/routes.py @@ -7,7 +7,22 @@ from pathlib import Path from typing import Any -from telegram_resender.whitelist import normalize_username +from telegram_resender.telegram_limits import fits_telegram_message +from telegram_resender.whitelist import normalize_username, parse_user_id + +_ROUTES_DOCUMENT_FIELDS = frozenset({"routes"}) +_ROUTE_FIELDS = frozenset( + { + "name", + "target_chat_id", + "allowed_user_ids", + "allowed_usernames", + "keywords_any", + "keywords_none", + "template", + "enabled", + } +) @dataclass(frozen=True, slots=True) @@ -21,12 +36,15 @@ class RouteRule: keywords_none: tuple[str, ...] template: str | None enabled: bool = True + allowed_user_ids: frozenset[int] = frozenset() - def matches(self, *, username: str | None, text: str) -> bool: + def matches(self, *, username: str | None, text: str, user_id: int | None = None) -> bool: """Return whether this route should receive the request.""" if not self.enabled: return False + if self.allowed_user_ids and user_id not in self.allowed_user_ids: + return False normalized_username = normalize_username(username) if self.allowed_usernames and normalized_username not in self.allowed_usernames: return False @@ -55,6 +73,7 @@ def default_route(target_chat_id: int) -> RouteRule: keywords_any=(), keywords_none=(), template=None, + allowed_user_ids=frozenset(), ) @@ -62,8 +81,16 @@ def load_routes(path: Path) -> tuple[RouteRule, ...]: """Load route rules from a JSON file.""" with path.open("r", encoding="utf-8") as stream: - payload = json.load(stream) - raw_routes = payload.get("routes", payload) if isinstance(payload, dict) else payload + payload = json.load(stream, object_pairs_hook=_reject_duplicate_keys) + if isinstance(payload, dict): + _reject_unknown_fields( + payload, + allowed_fields=_ROUTES_DOCUMENT_FIELDS, + context="Routes config", + ) + raw_routes = payload.get("routes") + else: + raw_routes = payload if not isinstance(raw_routes, list): msg = "Routes config must be a JSON list or an object with a 'routes' list" raise ValueError(msg) @@ -78,43 +105,171 @@ def _parse_route(item: Any, index: int) -> RouteRule: if not isinstance(item, dict): msg = f"Route #{index} must be an object" raise ValueError(msg) - name = str(item.get("name") or f"route-{index}") + _reject_unknown_fields(item, allowed_fields=_ROUTE_FIELDS, context=f"Route #{index}") + name = _parse_name(item.get("name"), index=index) if "target_chat_id" not in item: msg = f"Route '{name}' is missing target_chat_id" raise ValueError(msg) return RouteRule( name=name, - target_chat_id=int(item["target_chat_id"]), + target_chat_id=_parse_target_chat_id(item["target_chat_id"], route_name=name), + allowed_user_ids=_normalize_user_ids(item.get("allowed_user_ids", ())), allowed_usernames=_normalize_usernames(item.get("allowed_usernames", ())), keywords_any=_normalize_keywords(item.get("keywords_any", ())), keywords_none=_normalize_keywords(item.get("keywords_none", ())), - template=_optional_string(item.get("template")), - enabled=bool(item.get("enabled", True)), + template=_parse_template(item.get("template"), route_name=name), + enabled=_parse_enabled(item.get("enabled", True), route_name=name), ) +def _parse_name(value: Any, *, index: int) -> str: + if value is None: + return f"route-{index}" + if not isinstance(value, str) or not value.strip(): + msg = f"Route #{index} name must be a non-empty string" + raise ValueError(msg) + return value.strip() + + +def _parse_target_chat_id(value: Any, *, route_name: str) -> int: + if isinstance(value, bool) or not isinstance(value, (int, str)): + msg = f"Route '{route_name}' target_chat_id must be a non-zero integer" + raise ValueError(msg) + try: + target_chat_id = int(value) + except ValueError as exc: + msg = f"Route '{route_name}' target_chat_id must be a non-zero integer" + raise ValueError(msg) from exc + if target_chat_id == 0: + msg = f"Route '{route_name}' target_chat_id must be a non-zero integer" + raise ValueError(msg) + return target_chat_id + + +def _parse_enabled(value: Any, *, route_name: str) -> bool: + if not isinstance(value, bool): + msg = f"Route '{route_name}' enabled must be a boolean" + raise ValueError(msg) + return value + + def _normalize_usernames(value: Any) -> frozenset[str]: - return frozenset( - username - for item in _as_sequence(value) - if (username := normalize_username(str(item))) is not None - ) + normalized: set[str] = set() + for item in _string_sequence(value, field_name="allowed_usernames"): + username = normalize_username(item) + if username is None: + msg = "Route allowed_usernames must contain valid non-empty usernames" + raise ValueError(msg) + normalized.add(username) + return frozenset(normalized) + + +def _normalize_user_ids(value: Any) -> frozenset[int]: + parsed: set[int] = set() + for item in _user_id_sequence(value): + try: + user_id = parse_user_id(item) + except ValueError as exc: + msg = "Route allowed_user_ids must contain positive integers" + raise ValueError(msg) from exc + if user_id is None: + msg = "Route allowed_user_ids must contain positive integers" + raise ValueError(msg) + parsed.add(user_id) + return frozenset(parsed) def _normalize_keywords(value: Any) -> tuple[str, ...]: - return tuple(str(item).strip().casefold() for item in _as_sequence(value) if str(item).strip()) + return tuple( + item.strip().casefold() + for item in _string_sequence(value, field_name="keyword filters") + if item.strip() + ) def _optional_string(value: Any) -> str | None: if value is None: return None - text = str(value) - return text or None + if not isinstance(value, str): + msg = "Route template must be a string or null" + raise ValueError(msg) + if value == "": + return None + if not value.strip(): + msg = "Route template must contain non-whitespace text" + raise ValueError(msg) + return value + +def _parse_template(value: Any, *, route_name: str) -> str | None: + template = _optional_string(value) + if template is None: + return None + try: + rendered_sample = template.format(route=route_name, request="request") + except (AttributeError, IndexError, KeyError, TypeError, ValueError) as exc: + msg = f"Route '{route_name}' has invalid template: {exc}" + raise ValueError(msg) from exc + if not fits_telegram_message(rendered_sample): + msg = f"Route '{route_name}' has invalid template: rendered text exceeds Telegram limits" + raise ValueError(msg) + return template -def _as_sequence(value: Any) -> tuple[Any, ...]: + +def _string_sequence(value: Any, *, field_name: str) -> tuple[str, ...]: if value is None: return () if isinstance(value, str): - return (value,) - return tuple(value) + values = (value,) + elif isinstance(value, (list, tuple)) and all(isinstance(item, str) for item in value): + values = tuple(value) + else: + msg = f"Route {field_name} must be a string, a list of strings, or null" + raise ValueError(msg) + if any(not item.strip() for item in values): + msg = f"Route {field_name} must contain only non-empty strings" + raise ValueError(msg) + return values + + +def _user_id_sequence(value: Any) -> tuple[int | str, ...]: + if value is None: + return () + if isinstance(value, bool): + msg = "Route allowed_user_ids must be an integer, a list of integers, or null" + raise ValueError(msg) + if isinstance(value, (int, str)): + values = (value,) + elif isinstance(value, (list, tuple)) and all( + isinstance(item, (int, str)) and not isinstance(item, bool) for item in value + ): + values = tuple(value) + else: + msg = "Route allowed_user_ids must be an integer, a list of integers, or null" + raise ValueError(msg) + return values + + +def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + msg = f"Routes config contains duplicate JSON key {key!r}" + raise ValueError(msg) + result[key] = value + return result + + +def _reject_unknown_fields( + item: dict[Any, Any], + *, + allowed_fields: frozenset[str], + context: str, +) -> None: + unknown_fields = sorted(repr(key) for key in item if key not in allowed_fields) + if not unknown_fields: + return + allowed = ", ".join(sorted(allowed_fields)) + unknown = ", ".join(unknown_fields) + msg = f"{context} contains unknown field(s) {unknown}; allowed fields: {allowed}" + raise ValueError(msg) diff --git a/src/telegram_resender/service.py b/src/telegram_resender/service.py index ad80686..6663f85 100644 --- a/src/telegram_resender/service.py +++ b/src/telegram_resender/service.py @@ -44,6 +44,11 @@ def whitelist_count(self) -> int: return len(self._whitelist.user_ids) + def is_authorized(self, user_id: int | None) -> bool: + """Return whether a Telegram user ID is authorized by the current whitelist.""" + + return user_id is not None and self._whitelist.contains(user_id) + def reload_whitelist(self, path: Path) -> int: """Reload whitelist from disk and return the new user count.""" @@ -91,19 +96,30 @@ def handle_text(self, message: IncomingMessage) -> ForwardingDecision: ) request_id = self._formatter.format_request_id(message) - matching_routes = tuple( - route - for route in self._routes - if route.matches(username=message.user.username, text=parsed_request.fields["building"]) - ) + legacy_request_id = self._formatter.format_legacy_request_id(message) + request_id_aliases = (legacy_request_id,) if legacy_request_id is not None else () + matching_routes_by_target: dict[int, RouteRule] = {} + for route in self._routes: + if route.matches( + username=message.user.username, + user_id=message.user.id, + text=parsed_request.fields["building"], + ): + matching_routes_by_target.setdefault(route.target_chat_id, route) + matching_routes = tuple(matching_routes_by_target.values()) if not matching_routes: return ForwardingDecision( should_forward=False, response_text=self._no_route_matched_message, reason="no_route_matched", request_id=request_id, + request_id_aliases=request_id_aliases, ) forward_text = self._formatter.format_forward(message) + forward_payloads = tuple( + (route.target_chat_id, route.render_forward_text(forward_text)) + for route in matching_routes + ) return ForwardingDecision( should_forward=True, response_text=self._request_accepted_message, @@ -111,6 +127,8 @@ def handle_text(self, message: IncomingMessage) -> ForwardingDecision: forward_text=forward_text, request_id=request_id, target_chat_ids=tuple(route.target_chat_id for route in matching_routes), + forward_payloads=forward_payloads, + request_id_aliases=request_id_aliases, ) def render_forward_text_for_target(self, target_chat_id: int, default_text: str) -> str: diff --git a/src/telegram_resender/settings.py b/src/telegram_resender/settings.py index 186702a..3217d75 100644 --- a/src/telegram_resender/settings.py +++ b/src/telegram_resender/settings.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from pathlib import Path from typing import Literal @@ -9,6 +10,12 @@ from pydantic_settings import BaseSettings, SettingsConfigDict from telegram_resender.messages import Locale, MessageCatalog, message_catalog +from telegram_resender.telegram_limits import ( + TELEGRAM_MESSAGE_MAX_UTF16_UNITS, + fits_telegram_message, +) + +_BOT_TOKEN_PATTERN = re.compile(r"[1-9][0-9]*:[A-Za-z0-9_-]{35}", re.ASCII) class Settings(BaseSettings): @@ -44,7 +51,7 @@ class Settings(BaseSettings): ) storage_path: Path = Field( default=Path("telegram_resender.sqlite3"), - description="SQLite request delivery log path.", + description="SQLite delivery, lease, and pending-confirmation state path.", validation_alias=AliasChoices("TELEGRAM_RESENDER_STORAGE_PATH", "STORAGE_PATH"), ) polling_timeout: int = Field(default=30, ge=1, le=120) @@ -52,6 +59,8 @@ class Settings(BaseSettings): log_format: Literal["TEXT", "JSON"] = "TEXT" locale: Locale = "ru" confirm_before_forward: bool = False + pending_request_ttl_seconds: int = Field(default=900, ge=30, le=86_400) + delivery_lease_seconds: int = Field(default=300, ge=30, le=3_600) delivery_max_attempts: int = Field(default=3, ge=1, le=10) delivery_retry_backoff: float = Field(default=1.0, ge=0.0, le=60.0) admin_ids_raw: str = Field( @@ -85,15 +94,72 @@ def validate_bot_token(cls, value: str) -> str: """Reject placeholder tokens before the application can start.""" token = value.strip() - placeholders = {"*****", "changeme", "change-me", "your-token", "your_bot_token"} + placeholders = { + "*****", + "changeme", + "change-me", + "replace-me", + "your-token", + "your_bot_token", + } if not token or token.lower() in placeholders or set(token) == {"*"}: msg = "bot_token must be a real Telegram Bot API token, not a placeholder" raise ValueError(msg) - if ":" not in token: - msg = "bot_token must look like a Telegram Bot API token and contain ':'" + _bot_id, _separator, secret = token.partition(":") + if secret.lower() in placeholders or (secret and set(secret) == {"*"}): + msg = "bot_token must be a real Telegram Bot API token, not a placeholder" + raise ValueError(msg) + if _BOT_TOKEN_PATTERN.fullmatch(token) is None: + msg = ( + "bot_token must look like a Telegram Bot API token: " + "positive numeric bot id, ':', and a 35-character ASCII secret" + ) raise ValueError(msg) return token + @field_validator("request_accepted_message", "access_denied_message") + @classmethod + def validate_configurable_message(cls, value: str | None) -> str | None: + """Reject configured replies that Telegram cannot send as one message.""" + + if value is None or value == "": + return value + if not value.strip(): + msg = "configured Telegram messages must contain non-whitespace text" + raise ValueError(msg) + if not fits_telegram_message(value): + msg = ( + "configured Telegram messages must not exceed " + f"{TELEGRAM_MESSAGE_MAX_UTF16_UNITS} UTF-16 code units" + ) + raise ValueError(msg) + return value + + @field_validator("forward_chat_id") + @classmethod + def validate_forward_chat_id(cls, value: int) -> int: + """Reject Telegram's reserved zero chat ID before delivery.""" + + if value == 0: + msg = "forward_chat_id must be a non-zero Telegram chat ID" + raise ValueError(msg) + return value + + @field_validator("admin_ids_raw") + @classmethod + def validate_admin_ids_raw(cls, value: str) -> str: + """Reject malformed admin IDs during settings validation, not command handling.""" + + try: + for part in value.split(","): + if part.strip(): + if int(part.strip()) <= 0: + raise ValueError + except ValueError as exc: + msg = "admin IDs must be comma-separated integers greater than zero" + raise ValueError(msg) from exc + return value + @field_validator("whitelist_path") @classmethod def normalize_whitelist_path(cls, value: Path) -> Path: diff --git a/src/telegram_resender/storage.py b/src/telegram_resender/storage.py index 1cc15a8..1ee2490 100644 --- a/src/telegram_resender/storage.py +++ b/src/telegram_resender/storage.py @@ -1,18 +1,60 @@ -"""SQLite request delivery log.""" +"""SQLite-backed delivery and pending-confirmation state.""" from __future__ import annotations import csv +import json import sqlite3 from collections.abc import Iterable, Iterator from contextlib import contextmanager from dataclasses import dataclass from datetime import UTC, date, datetime from pathlib import Path -from typing import Literal, TextIO +from secrets import token_hex +from time import time +from typing import Any, Literal, TextIO DeliveryStatus = Literal["pending", "delivered", "failed", "skipped"] +PendingRequestKey = tuple[int, int | None] _DANGEROUS_CSV_PREFIXES = ("=", "+", "-", "@", "\t", "\r", "\n") +_MAX_STORED_ERROR_LENGTH = 2_000 + +_EXPECTED_DELIVERY_COLUMNS = ( + ("request_id", "TEXT", True), + ("target_chat_id", "INTEGER", True), + ("sender_username", "TEXT", False), + ("created_at", "TEXT", True), + ("validation_status", "TEXT", True), + ("delivery_status", "TEXT", True), + ("last_error", "TEXT", False), +) +_EXPECTED_DELIVERY_UNIQUE_INDEX = ("request_id", "target_chat_id") +_EXPECTED_DELIVERY_LEASE_COLUMNS = ( + ("request_id", "TEXT", True), + ("target_chat_id", "INTEGER", True), + ("owner_token", "TEXT", True), + ("version", "INTEGER", True), + ("lease_expires_at", "REAL", True), +) +_EXPECTED_PENDING_SEQUENCE_COLUMNS = ( + ("chat_id", "INTEGER", True), + ("user_id", "INTEGER", True), + ("current_version", "INTEGER", True), +) +_EXPECTED_PENDING_REQUEST_COLUMNS = ( + ("chat_id", "INTEGER", True), + ("user_id", "INTEGER", True), + ("version", "INTEGER", True), + ("request_id", "TEXT", True), + ("request_id_aliases", "TEXT", True), + ("forward_payloads", "TEXT", True), + ("sender_user_id", "INTEGER", True), + ("sender_username", "TEXT", False), + ("expires_at", "REAL", True), + ("owner_token", "TEXT", False), + ("owner_version", "INTEGER", True), + ("owner_expires_at", "REAL", False), +) @dataclass(frozen=True, slots=True) @@ -28,13 +70,49 @@ class DeliveryRecord: last_error: str | None +@dataclass(frozen=True, slots=True) +class DeliveryLease: + """Exclusive, versioned ownership of one request/target delivery.""" + + request_id: str + request_id_aliases: tuple[str, ...] + target_chat_id: int + owner_token: str + version: int + expires_at: float + + +class DeliveryInProgressError(RuntimeError): + """Raised when another live owner already holds the delivery lease.""" + + +@dataclass(frozen=True, slots=True) +class PendingRequest: + """Persisted forwarding payload waiting for explicit confirmation.""" + + version: int + request_id: str + request_id_aliases: tuple[str, ...] + forward_text_by_chat_id: tuple[tuple[int, str], ...] + sender_user_id: int + sender_username: str | None + expires_at: float + owner_token: str | None = None + owner_version: int = 0 + owner_expires_at: float | None = None + + class RequestLog: - """Small SQLite-backed request delivery log.""" + """SQLite delivery log with atomic, expiring ownership leases.""" - def __init__(self, path: Path) -> None: + def __init__(self, path: Path, *, lease_seconds: int = 300) -> None: + if lease_seconds <= 0: + msg = "delivery lease duration must be positive" + raise ValueError(msg) self._path = path + self._lease_seconds = lease_seconds self._path.parent.mkdir(parents=True, exist_ok=True) - self._ensure_schema() + _ensure_storage_schema(self._path) def begin_delivery( self, @@ -42,17 +120,49 @@ def begin_delivery( request_id: str, target_chat_id: int, sender_username: str | None, - ) -> bool: - """Create or update a delivery row. + request_id_aliases: tuple[str, ...] = (), + ) -> DeliveryLease | None: + """Atomically claim a request/target pair unless delivered or actively leased.""" - Returns False when this request/target pair has already been delivered. - """ - - existing = self._get_status(request_id, target_chat_id) - if existing == "delivered": - return False - now = _utc_now() + request_ids = _request_ids(request_id, request_id_aliases) + now = time() + owner_token = token_hex(16) with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing_statuses = _get_statuses(connection, request_ids, target_chat_id) + if "delivered" in existing_statuses.values(): + return None + + placeholders = ", ".join("?" for _ in request_ids) + claim_rows = connection.execute( + f""" + SELECT version, lease_expires_at + FROM delivery_leases + WHERE request_id IN ({placeholders}) AND target_chat_id = ? + """, + (*request_ids, target_chat_id), + ).fetchall() + if any(float(row[1]) > now for row in claim_rows): + msg = f"request {request_id!r} is already being delivered to {target_chat_id}" + raise DeliveryInProgressError(msg) + + version = max((int(row[0]) for row in claim_rows), default=0) + 1 + expires_at = now + self._lease_seconds + for claimed_request_id in request_ids: + connection.execute( + """ + INSERT INTO delivery_leases ( + request_id, target_chat_id, owner_token, version, lease_expires_at + ) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(request_id, target_chat_id) DO UPDATE SET + owner_token = excluded.owner_token, + version = excluded.version, + lease_expires_at = excluded.lease_expires_at + """, + (claimed_request_id, target_chat_id, owner_token, version, expires_at), + ) + connection.execute( """ INSERT INTO request_deliveries ( @@ -64,29 +174,97 @@ def begin_delivery( delivery_status = 'pending', last_error = NULL """, - (request_id, target_chat_id, sender_username, now), + (request_id, target_chat_id, sender_username, _utc_now()), ) - return True + return DeliveryLease( + request_id=request_id, + request_id_aliases=tuple(item for item in request_ids if item != request_id), + target_chat_id=target_chat_id, + owner_token=owner_token, + version=version, + expires_at=expires_at, + ) def mark_delivery( self, *, - request_id: str, - target_chat_id: int, + lease: DeliveryLease, status: DeliveryStatus, error: str | None = None, - ) -> None: - """Persist delivery result for a request/target pair.""" + ) -> bool: + """Persist a result only while the supplied owner/version still owns the lease.""" + request_ids = _request_ids(lease.request_id, lease.request_id_aliases) with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + if not _owns_delivery_lease( + connection, + request_ids=request_ids, + target_chat_id=lease.target_chat_id, + owner_token=lease.owner_token, + version=lease.version, + ): + return False + connection.execute( """ UPDATE request_deliveries SET delivery_status = ?, last_error = ? WHERE request_id = ? AND target_chat_id = ? """, - (status, error, request_id, target_chat_id), + ( + status, + error[:_MAX_STORED_ERROR_LENGTH] if error is not None else None, + lease.request_id, + lease.target_chat_id, + ), ) + placeholders = ", ".join("?" for _ in request_ids) + connection.execute( + f""" + UPDATE delivery_leases + SET lease_expires_at = 0 + WHERE request_id IN ({placeholders}) + AND target_chat_id = ? + AND owner_token = ? + AND version = ? + """, + (*request_ids, lease.target_chat_id, lease.owner_token, lease.version), + ) + return True + + def renew_delivery(self, lease: DeliveryLease, *, wait_seconds: float = 0.0) -> bool: + """Extend a live lease across a known retry wait without changing ownership.""" + + request_ids = _request_ids(lease.request_id, lease.request_id_aliases) + new_expires_at = time() + self._lease_seconds + max(wait_seconds, 0.0) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + if not _owns_delivery_lease( + connection, + request_ids=request_ids, + target_chat_id=lease.target_chat_id, + owner_token=lease.owner_token, + version=lease.version, + ): + return False + placeholders = ", ".join("?" for _ in request_ids) + connection.execute( + f""" + UPDATE delivery_leases + SET lease_expires_at = ? + WHERE request_id IN ({placeholders}) + AND target_chat_id = ? AND owner_token = ? AND version = ? + """, + ( + new_expires_at, + *request_ids, + lease.target_chat_id, + lease.owner_token, + lease.version, + ), + ) + return True def records_since(self, since: date) -> tuple[DeliveryRecord, ...]: """Return delivery records created at or after the given date.""" @@ -106,47 +284,348 @@ def records_since(self, since: date) -> tuple[DeliveryRecord, ...]: return tuple(DeliveryRecord(*row) for row in rows) def check(self) -> None: - """Open the database and ensure the schema exists.""" + """Open the database and validate every runtime storage table.""" - self._ensure_schema() + _ensure_storage_schema(self._path) - def _get_status(self, request_id: str, target_chat_id: int) -> str | None: - with self._connect() as connection: - row = connection.execute( + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: + with _connect(self._path) as connection: + yield connection + + +class PendingRequestStore: + """Persistent pending confirmations with TTL and versioned claim ownership.""" + + def __init__(self, path: Path, *, ttl_seconds: int, claim_seconds: int = 300) -> None: + if ttl_seconds <= 0 or claim_seconds <= 0: + msg = "pending TTL and claim duration must be positive" + raise ValueError(msg) + self._path = path + self._ttl_seconds = ttl_seconds + self._claim_seconds = claim_seconds + self._path.parent.mkdir(parents=True, exist_ok=True) + _ensure_storage_schema(self._path) + + def publish( + self, + key: PendingRequestKey, + *, + request_id: str, + request_id_aliases: tuple[str, ...], + forward_text_by_chat_id: tuple[tuple[int, str], ...], + sender_user_id: int | None, + sender_username: str | None, + ) -> PendingRequest: + """Publish a visible preview, replacing only an unclaimed duplicate ID.""" + + chat_id, user_id = _require_pending_key(key) + if sender_user_id != user_id: + msg = "pending request owner must match its storage key" + raise ValueError(msg) + now = time() + with _connect(self._path) as connection: + connection.execute("BEGIN IMMEDIATE") + _purge_pending(connection, now) + sequence_row = connection.execute( """ - SELECT delivery_status - FROM request_deliveries - WHERE request_id = ? AND target_chat_id = ? + SELECT current_version + FROM pending_request_sequences + WHERE chat_id = ? AND user_id = ? """, - (request_id, target_chat_id), + (chat_id, user_id), ).fetchone() - return str(row[0]) if row else None - - def _ensure_schema(self) -> None: - with self._connect() as connection: + version = (int(sequence_row[0]) if sequence_row else 0) + 1 + connection.execute( + """ + INSERT INTO pending_request_sequences (chat_id, user_id, current_version) + VALUES (?, ?, ?) + ON CONFLICT(chat_id, user_id) DO UPDATE SET + current_version = excluded.current_version + """, + (chat_id, user_id, version), + ) + connection.execute( + """ + DELETE FROM pending_requests + WHERE chat_id = ? AND user_id = ? AND request_id = ? + AND (owner_token IS NULL OR owner_expires_at <= ?) + """, + (chat_id, user_id, request_id, now), + ) + expires_at = now + self._ttl_seconds connection.execute( """ - CREATE TABLE IF NOT EXISTS request_deliveries ( - request_id TEXT NOT NULL, - target_chat_id INTEGER NOT NULL, - sender_username TEXT, - created_at TEXT NOT NULL, - validation_status TEXT NOT NULL, - delivery_status TEXT NOT NULL, - last_error TEXT, - PRIMARY KEY (request_id, target_chat_id) + INSERT INTO pending_requests ( + chat_id, user_id, version, request_id, request_id_aliases, + forward_payloads, sender_user_id, sender_username, expires_at, + owner_token, owner_version, owner_expires_at ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 0, NULL) + """, + ( + chat_id, + user_id, + version, + request_id, + json.dumps(request_id_aliases, ensure_ascii=False), + json.dumps(forward_text_by_chat_id, ensure_ascii=False), + sender_user_id, + sender_username, + expires_at, + ), + ) + return PendingRequest( + version=version, + request_id=request_id, + request_id_aliases=request_id_aliases, + forward_text_by_chat_id=forward_text_by_chat_id, + sender_user_id=user_id, + sender_username=sender_username, + expires_at=expires_at, + ) + + def claim( + self, + key: PendingRequestKey, + *, + request_id: str | None = None, + ) -> PendingRequest | None: + """Atomically claim the latest matching visible request.""" + + resolved_key = _optional_pending_key(key) + if resolved_key is None: + return None + chat_id, user_id = resolved_key + now = time() + owner_token = token_hex(16) + with _connect(self._path) as connection: + connection.execute("BEGIN IMMEDIATE") + _prepare_pending_key(connection, chat_id=chat_id, user_id=user_id, now=now) + parameters: list[object] = [chat_id, user_id, now] + request_predicate = "" + if request_id is not None: + request_predicate = "AND request_id = ?" + parameters.append(request_id) + row = connection.execute( + f""" + SELECT version, request_id, request_id_aliases, forward_payloads, + sender_user_id, sender_username, expires_at, + owner_token, owner_version, owner_expires_at + FROM pending_requests + WHERE chat_id = ? AND user_id = ? AND expires_at > ? + AND owner_token IS NULL {request_predicate} + ORDER BY version DESC + LIMIT 1 + """, + tuple(parameters), + ).fetchone() + if row is None: + return None + version = int(row[0]) + owner_version = int(row[8]) + 1 + owner_expires_at = now + self._claim_seconds + updated = connection.execute( """ + UPDATE pending_requests + SET owner_token = ?, owner_version = ?, owner_expires_at = ? + WHERE chat_id = ? AND user_id = ? AND version = ? AND owner_token IS NULL + """, + ( + owner_token, + owner_version, + owner_expires_at, + chat_id, + user_id, + version, + ), ) + if updated.rowcount != 1: + return None + claimed_row = (*row[:7], owner_token, owner_version, owner_expires_at) + return _pending_from_row(claimed_row) - @contextmanager - def _connect(self) -> Iterator[sqlite3.Connection]: - connection = sqlite3.connect(self._path) - try: - with connection: - yield connection - finally: - connection.close() + def release(self, key: PendingRequestKey, pending: PendingRequest) -> bool: + """Release a failed claim for retry if its original preview TTL remains valid.""" + + resolved_key = _optional_pending_key(key) + if resolved_key is None or pending.owner_token is None: + return False + chat_id, user_id = resolved_key + now = time() + with _connect(self._path) as connection: + connection.execute("BEGIN IMMEDIATE") + if pending.expires_at <= now: + connection.execute( + """ + DELETE FROM pending_requests + WHERE chat_id = ? AND user_id = ? AND version = ? + AND owner_token = ? AND owner_version = ? + """, + ( + chat_id, + user_id, + pending.version, + pending.owner_token, + pending.owner_version, + ), + ) + return False + updated = connection.execute( + """ + UPDATE pending_requests + SET owner_token = NULL, owner_expires_at = NULL + WHERE chat_id = ? AND user_id = ? AND version = ? + AND owner_token = ? AND owner_version = ? + """, + ( + chat_id, + user_id, + pending.version, + pending.owner_token, + pending.owner_version, + ), + ) + if updated.rowcount == 1: + newest_row = connection.execute( + """ + SELECT MAX(version) + FROM pending_requests + WHERE chat_id = ? AND user_id = ? AND request_id = ? + AND owner_token IS NULL AND expires_at > ? + """, + (chat_id, user_id, pending.request_id, now), + ).fetchone() + newest_version = int(newest_row[0]) if newest_row and newest_row[0] else None + if newest_version is not None: + connection.execute( + """ + DELETE FROM pending_requests + WHERE chat_id = ? AND user_id = ? AND request_id = ? + AND owner_token IS NULL AND version <> ? + """, + (chat_id, user_id, pending.request_id, newest_version), + ) + return updated.rowcount == 1 + + def complete(self, key: PendingRequestKey, pending: PendingRequest) -> bool: + """Delete a successfully handled request only for its current owner/version.""" + + resolved_key = _optional_pending_key(key) + if resolved_key is None or pending.owner_token is None: + return False + chat_id, user_id = resolved_key + with _connect(self._path) as connection: + connection.execute("BEGIN IMMEDIATE") + deleted = connection.execute( + """ + DELETE FROM pending_requests + WHERE chat_id = ? AND user_id = ? AND version = ? + AND owner_token = ? AND owner_version = ? + """, + ( + chat_id, + user_id, + pending.version, + pending.owner_token, + pending.owner_version, + ), + ) + return deleted.rowcount == 1 + + def renew( + self, + key: PendingRequestKey, + pending: PendingRequest, + *, + wait_seconds: float = 0.0, + ) -> bool: + """Extend a live confirmation claim across a known delivery retry wait.""" + + resolved_key = _optional_pending_key(key) + if resolved_key is None or pending.owner_token is None: + return False + chat_id, user_id = resolved_key + owner_expires_at = time() + self._claim_seconds + max(wait_seconds, 0.0) + with _connect(self._path) as connection: + connection.execute("BEGIN IMMEDIATE") + updated = connection.execute( + """ + UPDATE pending_requests + SET owner_expires_at = ? + WHERE chat_id = ? AND user_id = ? AND version = ? + AND owner_token = ? AND owner_version = ? + """, + ( + owner_expires_at, + chat_id, + user_id, + pending.version, + pending.owner_token, + pending.owner_version, + ), + ) + return updated.rowcount == 1 + + def cancel( + self, + key: PendingRequestKey, + *, + request_id: str | None = None, + ) -> PendingRequest | None: + """Atomically remove the latest matching request unless it is actively claimed.""" + + resolved_key = _optional_pending_key(key) + if resolved_key is None: + return None + chat_id, user_id = resolved_key + now = time() + with _connect(self._path) as connection: + connection.execute("BEGIN IMMEDIATE") + _prepare_pending_key(connection, chat_id=chat_id, user_id=user_id, now=now) + parameters: list[object] = [chat_id, user_id, now] + request_predicate = "" + if request_id is not None: + request_predicate = "AND request_id = ?" + parameters.append(request_id) + row = connection.execute( + f""" + SELECT version, request_id, request_id_aliases, forward_payloads, + sender_user_id, sender_username, expires_at, + owner_token, owner_version, owner_expires_at + FROM pending_requests + WHERE chat_id = ? AND user_id = ? AND expires_at > ? + AND owner_token IS NULL {request_predicate} + ORDER BY version DESC + LIMIT 1 + """, + tuple(parameters), + ).fetchone() + if row is None: + return None + pending = _pending_from_row(row) + deleted = connection.execute( + """ + DELETE FROM pending_requests + WHERE chat_id = ? AND user_id = ? AND version = ? AND owner_token IS NULL + """, + (chat_id, user_id, pending.version), + ) + if deleted.rowcount != 1: + return None + return pending + + def discard_all(self, key: PendingRequestKey) -> None: + """Remove every pending request for a user/chat pair.""" + + resolved_key = _optional_pending_key(key) + if resolved_key is None: + return + with _connect(self._path) as connection: + connection.execute( + "DELETE FROM pending_requests WHERE chat_id = ? AND user_id = ?", + resolved_key, + ) def export_records_csv(records: Iterable[DeliveryRecord], stream: TextIO) -> None: @@ -178,6 +657,139 @@ def export_records_csv(records: Iterable[DeliveryRecord], stream: TextIO) -> Non ) +def _request_ids(request_id: str, request_id_aliases: tuple[str, ...]) -> tuple[str, ...]: + return tuple(dict.fromkeys((request_id, *request_id_aliases))) + + +def _get_statuses( + connection: sqlite3.Connection, + request_ids: tuple[str, ...], + target_chat_id: int, +) -> dict[str, str]: + placeholders = ", ".join("?" for _ in request_ids) + rows = connection.execute( + f""" + SELECT request_id, delivery_status + FROM request_deliveries + WHERE request_id IN ({placeholders}) AND target_chat_id = ? + """, + (*request_ids, target_chat_id), + ).fetchall() + return {str(row[0]): str(row[1]) for row in rows} + + +def _owns_delivery_lease( + connection: sqlite3.Connection, + *, + request_ids: tuple[str, ...], + target_chat_id: int, + owner_token: str, + version: int, +) -> bool: + """Require one owner/version across the primary request ID and every alias.""" + + placeholders = ", ".join("?" for _ in request_ids) + rows = connection.execute( + f""" + SELECT request_id, owner_token, version + FROM delivery_leases + WHERE request_id IN ({placeholders}) AND target_chat_id = ? + """, + (*request_ids, target_chat_id), + ).fetchall() + expected_owner = (owner_token, version) + return len(rows) == len(request_ids) and all( + (str(row[1]), int(row[2])) == expected_owner for row in rows + ) + + +def _pending_from_row(row: tuple[Any, ...]) -> PendingRequest: + try: + aliases_value = json.loads(str(row[2])) + payloads_value = json.loads(str(row[3])) + if not isinstance(aliases_value, list) or not all( + isinstance(item, str) for item in aliases_value + ): + raise ValueError + if not isinstance(payloads_value, list) or not all( + isinstance(item, list) + and len(item) == 2 + and isinstance(item[0], int) + and not isinstance(item[0], bool) + and isinstance(item[1], str) + for item in payloads_value + ): + raise ValueError + return PendingRequest( + version=int(row[0]), + request_id=str(row[1]), + request_id_aliases=tuple(aliases_value), + forward_text_by_chat_id=tuple((int(item[0]), str(item[1])) for item in payloads_value), + sender_user_id=int(row[4]), + sender_username=str(row[5]) if row[5] is not None else None, + expires_at=float(row[6]), + owner_token=str(row[7]) if row[7] is not None else None, + owner_version=int(row[8]), + owner_expires_at=float(row[9]) if row[9] is not None else None, + ) + except (TypeError, ValueError, json.JSONDecodeError) as exc: + msg = "pending_requests contains invalid serialized data" + raise sqlite3.DatabaseError(msg) from exc + + +def _require_pending_key(key: PendingRequestKey) -> tuple[int, int]: + resolved = _optional_pending_key(key) + if resolved is None: + msg = "pending requests require a Telegram user ID" + raise ValueError(msg) + return resolved + + +def _optional_pending_key(key: PendingRequestKey) -> tuple[int, int] | None: + chat_id, user_id = key + return (chat_id, user_id) if user_id is not None else None + + +def _purge_pending(connection: sqlite3.Connection, now: float) -> None: + connection.execute( + """ + UPDATE pending_requests + SET owner_token = NULL, owner_expires_at = NULL + WHERE owner_token IS NOT NULL AND owner_expires_at <= ? + """, + (now,), + ) + connection.execute( + "DELETE FROM pending_requests WHERE expires_at <= ? AND owner_token IS NULL", + (now,), + ) + + +def _prepare_pending_key( + connection: sqlite3.Connection, + *, + chat_id: int, + user_id: int, + now: float, +) -> None: + connection.execute( + """ + UPDATE pending_requests + SET owner_token = NULL, owner_expires_at = NULL + WHERE chat_id = ? AND user_id = ? + AND owner_token IS NOT NULL AND owner_expires_at <= ? + """, + (chat_id, user_id, now), + ) + connection.execute( + """ + DELETE FROM pending_requests + WHERE chat_id = ? AND user_id = ? AND expires_at <= ? AND owner_token IS NULL + """, + (chat_id, user_id, now), + ) + + def _safe_csv_cell(value: str | None) -> str: if not value: return "" @@ -188,3 +800,146 @@ def _safe_csv_cell(value: str | None) -> str: def _utc_now() -> str: return datetime.now(UTC).isoformat() + + +@contextmanager +def _connect(path: Path) -> Iterator[sqlite3.Connection]: + connection = sqlite3.connect(path) + try: + with connection: + yield connection + finally: + connection.close() + + +def _ensure_storage_schema(path: Path) -> None: + with _connect(path) as connection: + connection.execute( + """ + CREATE TABLE IF NOT EXISTS request_deliveries ( + request_id TEXT NOT NULL, + target_chat_id INTEGER NOT NULL, + sender_username TEXT, + created_at TEXT NOT NULL, + validation_status TEXT NOT NULL, + delivery_status TEXT NOT NULL, + last_error TEXT, + PRIMARY KEY (request_id, target_chat_id) + ) + """ + ) + connection.execute( + """ + CREATE TABLE IF NOT EXISTS delivery_leases ( + request_id TEXT NOT NULL, + target_chat_id INTEGER NOT NULL, + owner_token TEXT NOT NULL, + version INTEGER NOT NULL, + lease_expires_at REAL NOT NULL, + PRIMARY KEY (request_id, target_chat_id) + ) + """ + ) + connection.execute( + """ + CREATE TABLE IF NOT EXISTS pending_request_sequences ( + chat_id INTEGER NOT NULL, + user_id INTEGER NOT NULL, + current_version INTEGER NOT NULL, + PRIMARY KEY (chat_id, user_id) + ) + """ + ) + connection.execute( + """ + CREATE TABLE IF NOT EXISTS pending_requests ( + chat_id INTEGER NOT NULL, + user_id INTEGER NOT NULL, + version INTEGER NOT NULL, + request_id TEXT NOT NULL, + request_id_aliases TEXT NOT NULL, + forward_payloads TEXT NOT NULL, + sender_user_id INTEGER NOT NULL, + sender_username TEXT, + expires_at REAL NOT NULL, + owner_token TEXT, + owner_version INTEGER NOT NULL, + owner_expires_at REAL, + PRIMARY KEY (chat_id, user_id, version) + ) + """ + ) + _validate_table_schema( + connection, + table="request_deliveries", + expected_columns=_EXPECTED_DELIVERY_COLUMNS, + expected_primary_key=_EXPECTED_DELIVERY_UNIQUE_INDEX, + ) + _validate_table_schema( + connection, + table="delivery_leases", + expected_columns=_EXPECTED_DELIVERY_LEASE_COLUMNS, + expected_primary_key=("request_id", "target_chat_id"), + ) + _validate_table_schema( + connection, + table="pending_request_sequences", + expected_columns=_EXPECTED_PENDING_SEQUENCE_COLUMNS, + expected_primary_key=("chat_id", "user_id"), + ) + _validate_table_schema( + connection, + table="pending_requests", + expected_columns=_EXPECTED_PENDING_REQUEST_COLUMNS, + expected_primary_key=("chat_id", "user_id", "version"), + ) + + +def _validate_delivery_schema(connection: sqlite3.Connection) -> None: + """Backward-compatible helper retained for focused schema tests.""" + + _validate_table_schema( + connection, + table="request_deliveries", + expected_columns=_EXPECTED_DELIVERY_COLUMNS, + expected_primary_key=_EXPECTED_DELIVERY_UNIQUE_INDEX, + ) + + +def _validate_table_schema( + connection: sqlite3.Connection, + *, + table: str, + expected_columns: tuple[tuple[str, str, bool], ...], + expected_primary_key: tuple[str, ...], +) -> None: + column_rows = connection.execute(f"PRAGMA table_info({table})").fetchall() + actual_columns = tuple((str(row[1]), str(row[2]).upper(), bool(row[3])) for row in column_rows) + if actual_columns != expected_columns: + msg = ( + f"{table} schema mismatch: expected columns " + f"{expected_columns!r}, got {actual_columns!r}" + ) + raise sqlite3.DatabaseError(msg) + + index_rows = connection.execute(f"PRAGMA index_list({table})").fetchall() + for index_row in index_rows: + index_name = str(index_row[1]) + is_unique = bool(index_row[2]) + is_primary_key = len(index_row) > 3 and str(index_row[3]) == "pk" + is_partial = bool(index_row[4]) if len(index_row) > 4 else False + if not is_unique or not is_primary_key or is_partial: + continue + indexed_columns = tuple( + str(row[0]) + for row in connection.execute( + "SELECT name FROM pragma_index_info(?) ORDER BY seqno", + (index_name,), + ).fetchall() + ) + if indexed_columns == expected_primary_key: + return + + key = ", ".join(expected_primary_key) + msg = f"{table} schema mismatch: missing primary key on ({key})" + raise sqlite3.DatabaseError(msg) diff --git a/src/telegram_resender/telegram_limits.py b/src/telegram_resender/telegram_limits.py new file mode 100644 index 0000000..bfc3e7b --- /dev/null +++ b/src/telegram_resender/telegram_limits.py @@ -0,0 +1,21 @@ +"""Telegram text length helpers shared by configuration and handlers.""" + +from __future__ import annotations + +TELEGRAM_MESSAGE_MAX_UTF16_UNITS = 4096 + + +def telegram_utf16_length(text: str) -> int: + """Return the length Telegram uses for text/entity offsets.""" + + return len(text.encode("utf-16-le")) // 2 + + +def fits_telegram_message(text: str) -> bool: + """Return whether text fits Telegram's sendMessage UTF-16 limit.""" + + try: + length = telegram_utf16_length(text) + except UnicodeEncodeError: + return False + return 1 <= length <= TELEGRAM_MESSAGE_MAX_UTF16_UNITS diff --git a/src/telegram_resender/whitelist.py b/src/telegram_resender/whitelist.py index e72a677..5f30c0d 100644 --- a/src/telegram_resender/whitelist.py +++ b/src/telegram_resender/whitelist.py @@ -12,16 +12,24 @@ def parse_user_id(value: int | str | None) -> int | None: if value is None: return None + if isinstance(value, bool): + msg = f"Telegram user id must be a positive integer: {value!r}" + raise ValueError(msg) if isinstance(value, int): - return value - stripped = value.strip() - if not stripped: - return None - try: - return int(stripped) - except ValueError as exc: - msg = f"Telegram user id must be an integer: {value!r}" - raise ValueError(msg) from exc + parsed = value + else: + stripped = value.strip() + if not stripped: + return None + try: + parsed = int(stripped) + except ValueError as exc: + msg = f"Telegram user id must be a positive integer: {value!r}" + raise ValueError(msg) from exc + if parsed <= 0: + msg = f"Telegram user id must be a positive integer: {value!r}" + raise ValueError(msg) + return parsed def normalize_username(username: str | None) -> str | None: diff --git a/tests/integration/test_bootstrap.py b/tests/integration/test_bootstrap.py index 27e1db2..a933f61 100644 --- a/tests/integration/test_bootstrap.py +++ b/tests/integration/test_bootstrap.py @@ -13,6 +13,7 @@ "Автомобиль: Ford Focus\n" "Госномер: А123ВС" ) +VALID_BOT_TOKEN = f"123456789:{'A' * 35}" def test_build_service_uses_project_files(tmp_path: Path) -> None: @@ -21,7 +22,7 @@ def test_build_service_uses_project_files(tmp_path: Path) -> None: whitelist_file = tmp_path / "whitelist.csv" whitelist_file.write_text("10\n", encoding="utf-8") settings = Settings( - bot_token="123:abc", + bot_token=VALID_BOT_TOKEN, forward_chat_id=100, whitelist_path=whitelist_file, ) @@ -36,3 +37,20 @@ def test_build_service_uses_project_files(tmp_path: Path) -> None: assert decision.should_forward is True assert decision.response_text == REQUEST_ACCEPTED_MESSAGE + assert len(decision.request_id_aliases) == 1 + assert decision.request_id_aliases[0].startswith("local-") + assert decision.request_id_aliases[0] != decision.request_id + + +def test_systemd_environment_paths_match_service_layout() -> None: + """The copyable systemd env must use the writable /opt data directory.""" + + root = Path(__file__).parents[2] + systemd_env = (root / ".env.systemd.example").read_text(encoding="utf-8") + service = (root / "deploy" / "telegram-resender.service").read_text(encoding="utf-8") + + assert "TELEGRAM_RESENDER_WHITELIST_PATH=/opt/telegram-resender/data/" in systemd_env + assert "TELEGRAM_RESENDER_STORAGE_PATH=/opt/telegram-resender/data/" in systemd_env + assert "EnvironmentFile=/opt/telegram-resender/.env.production" in service + assert "ReadWritePaths=/opt/telegram-resender/data" in service + assert "TELEGRAM_RESENDER_WHITELIST_PATH=/data/" not in systemd_env diff --git a/tests/security/test_settings_security.py b/tests/security/test_settings_security.py index 9d8affa..05d422c 100644 --- a/tests/security/test_settings_security.py +++ b/tests/security/test_settings_security.py @@ -4,6 +4,8 @@ from telegram_resender.settings import Settings +VALID_BOT_TOKEN = f"123456789:{'A' * 35}" + def test_placeholder_token_is_rejected() -> None: """Refuse known placeholder values before startup.""" @@ -18,11 +20,101 @@ def test_bot_token_shape_is_validated() -> None: with pytest.raises(ValueError, match="must look like a Telegram Bot API token"): Settings(bot_token="not-a-token", forward_chat_id=1) + with pytest.raises(ValueError, match="must look like a Telegram Bot API token"): + Settings(bot_token=":", forward_chat_id=1) + + for token in ("123::", "123:a:b", "123:short", f"123:{'ю' * 35}"): + with pytest.raises(ValueError, match="must look like a Telegram Bot API token"): + Settings(bot_token=token, forward_chat_id=1) + + assert Settings(bot_token=VALID_BOT_TOKEN, forward_chat_id=1).bot_token == VALID_BOT_TOKEN + + +def test_production_example_token_is_rejected_as_placeholder() -> None: + """The copyable production example must not pass as a real credential.""" + + with pytest.raises(ValueError, match="placeholder"): + Settings(bot_token="123456:replace-me", forward_chat_id=1) + + +def test_zero_forward_chat_id_is_rejected() -> None: + """Telegram chat ID zero can never be a valid delivery target.""" + + with pytest.raises(ValueError): + Settings(bot_token=VALID_BOT_TOKEN, forward_chat_id=0) + def test_admin_ids_are_parsed_from_comma_separated_env(monkeypatch: pytest.MonkeyPatch) -> None: """Admin IDs should be easy to configure from .env files.""" monkeypatch.setenv("TELEGRAM_RESENDER_ADMIN_IDS", "100, 200") - settings = Settings(bot_token="123:abc", forward_chat_id=1) + settings = Settings(bot_token=VALID_BOT_TOKEN, forward_chat_id=1) assert settings.admin_ids == frozenset({100, 200}) + + +def test_malformed_admin_ids_are_rejected_during_settings_validation() -> None: + """A bad admin ID must not remain latent until an admin command is handled.""" + + with pytest.raises(ValueError, match="comma-separated integers"): + Settings( + bot_token=VALID_BOT_TOKEN, + forward_chat_id=1, + admin_ids_raw="100, not-an-id", + ) + + with pytest.raises(ValueError, match="greater than zero"): + Settings(bot_token=VALID_BOT_TOKEN, forward_chat_id=1, admin_ids_raw="100, -1") + + +def test_pending_request_ttl_is_bounded() -> None: + """Pending state must have a positive, operationally bounded lifetime.""" + + with pytest.raises(ValueError): + Settings( + bot_token=VALID_BOT_TOKEN, + forward_chat_id=1, + pending_request_ttl_seconds=29, + ) + with pytest.raises(ValueError): + Settings( + bot_token=VALID_BOT_TOKEN, + forward_chat_id=1, + pending_request_ttl_seconds=86_401, + ) + + +def test_delivery_lease_is_bounded() -> None: + """Lease recovery must be neither immediate nor indefinitely stuck.""" + + with pytest.raises(ValueError): + Settings(bot_token=VALID_BOT_TOKEN, forward_chat_id=1, delivery_lease_seconds=29) + with pytest.raises(ValueError): + Settings(bot_token=VALID_BOT_TOKEN, forward_chat_id=1, delivery_lease_seconds=3_601) + + +def test_configurable_replies_use_telegram_utf16_limit() -> None: + """Astral Unicode must count as two UTF-16 units for configurable replies.""" + + accepted = "😀" * 2048 + settings = Settings( + bot_token=VALID_BOT_TOKEN, + forward_chat_id=1, + request_accepted_message=accepted, + access_denied_message=accepted, + ) + assert settings.request_accepted_message == accepted + assert settings.access_denied_message == accepted + + with pytest.raises(ValueError, match="4096 UTF-16"): + Settings( + bot_token=VALID_BOT_TOKEN, + forward_chat_id=1, + request_accepted_message="😀" * 2049, + ) + with pytest.raises(ValueError, match="4096 UTF-16"): + Settings( + bot_token=VALID_BOT_TOKEN, + forward_chat_id=1, + access_denied_message="x" * 4097, + ) diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index 7fa1ed0..70e0325 100644 --- a/tests/unit/test_app.py +++ b/tests/unit/test_app.py @@ -1,5 +1,6 @@ """Unit tests for aiogram adapter wiring.""" +import asyncio from collections.abc import Awaitable, Callable from datetime import UTC, datetime from pathlib import Path @@ -9,12 +10,14 @@ import pytest from aiogram import Dispatcher, Router +import telegram_resender.storage as storage_module from telegram_resender.app import ( build_service, create_dispatcher, create_router, incoming_from_message, ) +from telegram_resender.formatting import MessageFormatter from telegram_resender.messages import ( CAR_MODE_MESSAGE, HELP_MESSAGE, @@ -23,6 +26,7 @@ START_MESSAGE, ) from telegram_resender.settings import Settings +from telegram_resender.storage import RequestLog VALID_REQUEST = ( "Объект/здание: Башня А\n" @@ -30,6 +34,7 @@ "Автомобиль: Ford Focus\n" "Госномер: А123ВС" ) +VALID_BOT_TOKEN = f"123456789:{'A' * 35}" class FakeBot: @@ -42,6 +47,54 @@ async def send_message(self, chat_id: int, text: str) -> None: self.sent_messages.append((chat_id, text)) +class FailOnceBot(FakeBot): + """Bot double that simulates one delivery failure before succeeding.""" + + def __init__(self) -> None: + super().__init__() + self.attempts = 0 + + async def send_message(self, chat_id: int, text: str) -> None: + self.attempts += 1 + if self.attempts == 1: + raise RuntimeError("temporary delivery failure") + await super().send_message(chat_id, text) + + +class BlockingFailOnceBot(FakeBot): + """Hold the first delivery open so another request can be published concurrently.""" + + def __init__(self) -> None: + super().__init__() + self.started = asyncio.Event() + self.release = asyncio.Event() + self.attempts = 0 + + async def send_message(self, chat_id: int, text: str) -> None: + self.attempts += 1 + if self.attempts == 1: + self.started.set() + await self.release.wait() + raise RuntimeError("temporary delivery failure") + await super().send_message(chat_id, text) + + +class BlockingBot(FakeBot): + """Hold a successful delivery open to expose duplicate in-flight claims.""" + + def __init__(self) -> None: + super().__init__() + self.started = asyncio.Event() + self.release = asyncio.Event() + self.attempts = 0 + + async def send_message(self, chat_id: int, text: str) -> None: + self.attempts += 1 + self.started.set() + await self.release.wait() + await super().send_message(chat_id, text) + + class FakeMessage: """Minimal message double exposing only the fields handlers read.""" @@ -52,10 +105,12 @@ def __init__( username: str | None = "alice", user_id: int | None = 10, bot: FakeBot | None = None, + chat_id: int = 100, + message_id: int | None = 55, ) -> None: - self.chat = SimpleNamespace(id=100) + self.chat = SimpleNamespace(id=chat_id) self.text = text - self.message_id = 55 + self.message_id = message_id self.date = datetime(2026, 6, 12, 10, 30, tzinfo=UTC) self.from_user = ( SimpleNamespace( @@ -78,18 +133,31 @@ async def answer(self, text: str) -> None: self.answers.append(text) +class FailingAnswerMessage(FakeMessage): + """Message double whose Telegram reply fails before a preview is published.""" + + async def answer(self, text: str) -> None: + raise RuntimeError("preview delivery failed") + + Handler = Callable[[Any], Awaitable[None]] -def _settings(tmp_path: Path, *, confirm_before_forward: bool = False) -> Settings: +def _settings( + tmp_path: Path, + *, + confirm_before_forward: bool = False, + pending_request_ttl_seconds: int = 900, +) -> Settings: whitelist_path = tmp_path / "whitelist.csv" whitelist_path.write_text("10\n", encoding="utf-8") return Settings( - bot_token="123:abc", + bot_token=VALID_BOT_TOKEN, forward_chat_id=200, whitelist_path=whitelist_path, storage_path=tmp_path / "requests.sqlite3", confirm_before_forward=confirm_before_forward, + pending_request_ttl_seconds=pending_request_ttl_seconds, admin_ids_raw="10", ) @@ -110,7 +178,7 @@ def _settings_with_routes(tmp_path: Path) -> Settings: encoding="utf-8", ) return Settings( - bot_token="123:abc", + bot_token=VALID_BOT_TOKEN, forward_chat_id=999, whitelist_path=whitelist_path, routes_path=routes_path, @@ -277,6 +345,58 @@ async def test_forward_text_skips_already_delivered_request(tmp_path: Path) -> N assert len(bot.sent_messages) == 1 +@pytest.mark.asyncio +async def test_concurrent_duplicate_request_has_one_delivery_owner(tmp_path: Path) -> None: + """Two overlapping handlers with one request ID must issue one Telegram send.""" + + settings = _settings(tmp_path) + router = create_router(settings, build_service(settings)) + handler = _message_handlers(router)["forward_text"] + bot = BlockingBot() + first_message = FakeMessage(bot=bot, message_id=55) + second_message = FakeMessage(bot=bot, message_id=55) + + async def run_first() -> None: + await handler(first_message) + + first = asyncio.create_task(run_first()) + await bot.started.wait() + await asyncio.wait_for(handler(second_message), timeout=2) + + assert bot.attempts == 1 + assert second_message.answers == [RU_MESSAGES.request_in_progress] + bot.release.set() + await first + assert len(bot.sent_messages) == 1 + + +@pytest.mark.asyncio +async def test_forward_text_skips_legacy_fallback_request_id(tmp_path: Path) -> None: + """A pre-upgrade local request ID must remain idempotent after the hash migration.""" + + settings = _settings(tmp_path) + bot = FakeBot() + message = FakeMessage(bot=bot, message_id=None) + legacy_request_id = MessageFormatter().format_legacy_request_id( + incoming_from_message(message) # type: ignore[arg-type] + ) + assert legacy_request_id is not None + request_log = RequestLog(settings.storage_path) + lease = request_log.begin_delivery( + request_id=legacy_request_id, + target_chat_id=settings.forward_chat_id, + sender_username="alice", + ) + assert lease is not None + request_log.mark_delivery(lease=lease, status="delivered") + router = create_router(settings, build_service(settings)) + + await _message_handlers(router)["forward_text"](message) + + assert bot.sent_messages == [] + assert message.answers == [REQUEST_ACCEPTED_MESSAGE] + + @pytest.mark.asyncio async def test_forward_text_uses_matching_routes(tmp_path: Path) -> None: """Routes config should override the default forward chat target.""" @@ -321,6 +441,265 @@ async def test_forward_text_can_require_confirmation(tmp_path: Path) -> None: assert "Request id: tg-100-55" in bot.sent_messages[0][1] +@pytest.mark.asyncio +async def test_pending_confirmation_survives_router_restart(tmp_path: Path) -> None: + """A visible preview must remain confirmable after process/router reconstruction.""" + + settings = _settings(tmp_path, confirm_before_forward=True) + first_handlers = _message_handlers(create_router(settings, build_service(settings))) + bot = FakeBot() + request = FakeMessage(bot=bot) + await first_handlers["forward_text"](request) + + restarted_handlers = _message_handlers(create_router(settings, build_service(settings))) + confirm = FakeMessage(text="/confirm tg-100-55", bot=bot) + await restarted_handlers["confirm_request"](confirm) + + assert confirm.answers == [RU_MESSAGES.request_confirmed.format(request_id="tg-100-55")] + assert len(bot.sent_messages) == 1 + + +@pytest.mark.asyncio +async def test_pending_cancel_survives_restart_and_removes_request(tmp_path: Path) -> None: + """Cancellation through a new store instance must persist for later processes.""" + + settings = _settings(tmp_path, confirm_before_forward=True) + first_handlers = _message_handlers(create_router(settings, build_service(settings))) + bot = FakeBot() + await first_handlers["forward_text"](FakeMessage(bot=bot)) + + restarted_handlers = _message_handlers(create_router(settings, build_service(settings))) + cancel = FakeMessage(text="/cancel tg-100-55", bot=bot) + await restarted_handlers["cancel_request"](cancel) + assert cancel.answers == [RU_MESSAGES.request_cancelled.format(request_id="tg-100-55")] + + final_handlers = _message_handlers(create_router(settings, build_service(settings))) + confirm = FakeMessage(text="/confirm tg-100-55", bot=bot) + await final_handlers["confirm_request"](confirm) + assert confirm.answers == [RU_MESSAGES.no_pending_request] + assert bot.sent_messages == [] + + +@pytest.mark.asyncio +async def test_forward_text_rejects_payload_over_telegram_limit(tmp_path: Path) -> None: + """Formatting overhead must not turn a valid incoming message into an API failure.""" + + settings = _settings(tmp_path) + router = create_router(settings, build_service(settings)) + handler = _message_handlers(router)["forward_text"] + bot = FakeBot() + text = f"{VALID_REQUEST}\nКомментарий: {'x' * 3900}" + assert len(text) <= 4096 + message = FakeMessage(text=text, bot=bot) + + await handler(message) + + assert message.answers == [RU_MESSAGES.request_too_long] + assert bot.sent_messages == [] + + +@pytest.mark.asyncio +async def test_confirmation_preview_over_telegram_limit_is_not_left_pending(tmp_path: Path) -> None: + """An oversized preview should be rejected without leaving a confirmable request.""" + + settings = _settings(tmp_path, confirm_before_forward=True) + router = create_router(settings, build_service(settings)) + handlers = _message_handlers(router) + bot = FakeBot() + text = f"{VALID_REQUEST}\nКомментарий: {'x' * 3800}" + message = FakeMessage(text=text, bot=bot) + + await handlers["forward_text"](message) + + assert message.answers == [RU_MESSAGES.request_too_long] + assert bot.sent_messages == [] + + confirm = FakeMessage(bot=bot) + await handlers["confirm_request"](confirm) + assert confirm.answers == [RU_MESSAGES.no_pending_request] + + +@pytest.mark.asyncio +async def test_pending_confirmation_is_isolated_per_user_in_shared_chat(tmp_path: Path) -> None: + """Another participant in a shared chat must not confirm someone else's request.""" + + settings = _settings(tmp_path, confirm_before_forward=True) + router = create_router(settings, build_service(settings)) + handlers = _message_handlers(router) + bot = FakeBot() + + await handlers["forward_text"](FakeMessage(user_id=10, bot=bot, chat_id=-1000)) + + other_user = FakeMessage(user_id=999, bot=bot, chat_id=-1000) + await handlers["confirm_request"](other_user) + + assert other_user.answers == [RU_MESSAGES.no_pending_request] + assert bot.sent_messages == [] + + owner = FakeMessage(user_id=10, bot=bot, chat_id=-1000) + await handlers["confirm_request"](owner) + + assert owner.answers == [RU_MESSAGES.request_confirmed.format(request_id="tg--1000-55")] + assert len(bot.sent_messages) == 1 + + +@pytest.mark.asyncio +async def test_pending_confirmation_is_retained_after_delivery_failure(tmp_path: Path) -> None: + """A failed confirmed delivery should remain available for a safe retry.""" + + settings = _settings(tmp_path, confirm_before_forward=True) + router = create_router(settings, build_service(settings)) + handlers = _message_handlers(router) + bot = FailOnceBot() + + await handlers["forward_text"](FakeMessage(bot=bot)) + + failed_confirm = FakeMessage(bot=bot) + with pytest.raises(RuntimeError, match="temporary delivery failure"): + await handlers["confirm_request"](failed_confirm) + + assert failed_confirm.answers == [ + RU_MESSAGES.request_confirmation_failed_retry.format(request_id="tg-100-55") + ] + + retry = FakeMessage(bot=bot) + await handlers["confirm_request"](retry) + + assert retry.answers == [RU_MESSAGES.request_confirmed.format(request_id="tg-100-55")] + assert len(bot.sent_messages) == 1 + + +@pytest.mark.asyncio +async def test_retry_guidance_failure_does_not_hide_delivery_error(tmp_path: Path) -> None: + """A failed guidance reply must not replace the original monitored delivery error.""" + + settings = _settings(tmp_path, confirm_before_forward=True) + handlers = _message_handlers(create_router(settings, build_service(settings))) + bot = FailOnceBot() + await handlers["forward_text"](FakeMessage(bot=bot)) + + with pytest.raises(RuntimeError, match="temporary delivery failure"): + await handlers["confirm_request"](FailingAnswerMessage(bot=bot)) + + +@pytest.mark.asyncio +async def test_confirm_rechecks_current_whitelist(tmp_path: Path) -> None: + """Removing a user from the runtime whitelist must revoke an old pending request.""" + + settings = _settings(tmp_path, confirm_before_forward=True) + service = build_service(settings) + router = create_router(settings, service) + handlers = _message_handlers(router) + bot = FakeBot() + + await handlers["forward_text"](FakeMessage(bot=bot)) + settings.whitelist_path.write_text("999\n", encoding="utf-8") + service.reload_whitelist(settings.whitelist_path) + + confirm = FakeMessage(text="/confirm tg-100-55", bot=bot) + await handlers["confirm_request"](confirm) + + assert confirm.answers == [settings.messages.access_denied_unknown] + assert bot.sent_messages == [] + + +@pytest.mark.asyncio +async def test_expired_pending_request_cannot_be_confirmed( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Pending confirmations should expire even while the bot process remains alive.""" + + now = [100.0] + monkeypatch.setattr(storage_module, "time", lambda: now[0]) + settings = _settings( + tmp_path, + confirm_before_forward=True, + pending_request_ttl_seconds=30, + ) + router = create_router(settings, build_service(settings)) + handlers = _message_handlers(router) + bot = FakeBot() + + await handlers["forward_text"](FakeMessage(bot=bot)) + now[0] += 31 + handlers = _message_handlers(create_router(settings, build_service(settings))) + confirm = FakeMessage(text="/confirm tg-100-55", bot=bot) + await handlers["confirm_request"](confirm) + + assert confirm.answers == [RU_MESSAGES.no_pending_request] + assert bot.sent_messages == [] + + +@pytest.mark.asyncio +async def test_failed_preview_does_not_replace_visible_pending_request(tmp_path: Path) -> None: + """A preview must be delivered before its request can replace confirmation state.""" + + settings = _settings(tmp_path, confirm_before_forward=True) + router = create_router(settings, build_service(settings)) + handlers = _message_handlers(router) + bot = FakeBot() + await handlers["forward_text"](FakeMessage(bot=bot, message_id=55)) + + hidden_request = FailingAnswerMessage( + text=f"{VALID_REQUEST}\nКомментарий: hidden", + bot=bot, + message_id=56, + ) + with pytest.raises(RuntimeError, match="preview delivery failed"): + await handlers["forward_text"](hidden_request) + + confirm = FakeMessage(text="/confirm", bot=bot) + await handlers["confirm_request"](confirm) + + assert confirm.answers == [RU_MESSAGES.request_confirmed.format(request_id="tg-100-55")] + assert len(bot.sent_messages) == 1 + assert "hidden" not in bot.sent_messages[0][1] + + +@pytest.mark.asyncio +async def test_failed_in_flight_request_and_newer_pending_request_are_both_retained( + tmp_path: Path, +) -> None: + """A delivery failure must restore A without overwriting concurrently published B.""" + + settings = _settings(tmp_path, confirm_before_forward=True) + router = create_router(settings, build_service(settings)) + handlers = _message_handlers(router) + bot = BlockingFailOnceBot() + await handlers["forward_text"](FakeMessage(bot=bot, message_id=55)) + + async def confirm_first_request() -> None: + await handlers["confirm_request"]( + FakeMessage(text="/confirm tg-100-55", bot=bot, message_id=100) + ) + + first_confirm = asyncio.create_task(confirm_first_request()) + await bot.started.wait() + await handlers["forward_text"]( + FakeMessage( + text=f"{VALID_REQUEST}\nКомментарий: newer", + bot=bot, + message_id=56, + ) + ) + bot.release.set() + + with pytest.raises(RuntimeError, match="temporary delivery failure"): + await first_confirm + + retry_first = FakeMessage(text="/confirm tg-100-55", bot=bot, message_id=101) + await handlers["confirm_request"](retry_first) + confirm_newer = FakeMessage(text="/confirm tg-100-56", bot=bot, message_id=102) + await handlers["confirm_request"](confirm_newer) + + assert retry_first.answers == [RU_MESSAGES.request_confirmed.format(request_id="tg-100-55")] + assert confirm_newer.answers == [RU_MESSAGES.request_confirmed.format(request_id="tg-100-56")] + assert len(bot.sent_messages) == 2 + assert "newer" not in bot.sent_messages[0][1] + assert "newer" in bot.sent_messages[1][1] + + @pytest.mark.asyncio async def test_confirmation_can_be_cancelled(tmp_path: Path) -> None: """Pending confirmation should be cancellable without forwarding.""" diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 1138101..0ed27c9 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -1,5 +1,6 @@ """Unit tests for the command-line entry point.""" +import sqlite3 from collections.abc import Coroutine from io import StringIO from pathlib import Path @@ -11,6 +12,8 @@ from telegram_resender.settings import Settings from telegram_resender.storage import RequestLog +VALID_BOT_TOKEN = f"123456789:{'A' * 35}" + def test_main_runs_polling(monkeypatch: Any) -> None: """The CLI should delegate to the async polling runner.""" @@ -59,7 +62,7 @@ def test_doctor_reports_loaded_configuration( whitelist_path = tmp_path / "whitelist.csv" whitelist_path.write_text("10\n20\n", encoding="utf-8") monkeypatch.chdir(tmp_path) - monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", "123:abc") + monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", VALID_BOT_TOKEN) monkeypatch.setenv("TELEGRAM_RESENDER_FORWARD_CHAT_ID", "100") monkeypatch.setenv("TELEGRAM_RESENDER_WHITELIST_PATH", str(whitelist_path)) monkeypatch.setenv("TELEGRAM_RESENDER_STORAGE_PATH", str(tmp_path / "requests.sqlite3")) @@ -74,6 +77,7 @@ def test_doctor_reports_loaded_configuration( assert "Storage check: OK" in stdout.getvalue() assert "Admin users: 0" in stdout.getvalue() assert "Confirm before forward: False" in stdout.getvalue() + assert "Pending request TTL: 900s" in stdout.getvalue() assert stderr.getvalue() == "" @@ -82,13 +86,14 @@ def test_export_requests_outputs_csv(tmp_path: Path, monkeypatch: pytest.MonkeyP storage_path = tmp_path / "requests.sqlite3" request_log = RequestLog(storage_path) - request_log.begin_delivery( + lease = request_log.begin_delivery( request_id="req-1", target_chat_id=100, sender_username="alice", ) - request_log.mark_delivery(request_id="req-1", target_chat_id=100, status="delivered") - monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", "123:abc") + assert lease is not None + request_log.mark_delivery(lease=lease, status="delivered") + monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", VALID_BOT_TOKEN) monkeypatch.setenv("TELEGRAM_RESENDER_FORWARD_CHAT_ID", "100") monkeypatch.setenv("TELEGRAM_RESENDER_STORAGE_PATH", str(storage_path)) stdout = StringIO() @@ -108,7 +113,7 @@ def test_health_reports_ok_for_valid_configuration( whitelist_path = tmp_path / "whitelist.csv" whitelist_path.write_text("10\n", encoding="utf-8") - monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", "123:abc") + monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", VALID_BOT_TOKEN) monkeypatch.setenv("TELEGRAM_RESENDER_FORWARD_CHAT_ID", "100") monkeypatch.setenv("TELEGRAM_RESENDER_WHITELIST_PATH", str(whitelist_path)) monkeypatch.setenv("TELEGRAM_RESENDER_STORAGE_PATH", str(tmp_path / "requests.sqlite3")) @@ -119,3 +124,69 @@ def test_health_reports_ok_for_valid_configuration( assert stdout.getvalue() == "health=ok\n" assert stderr.getvalue() == "" + + +def test_health_reports_sqlite_errors_without_traceback( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An unusable storage path should produce a normal health-check failure.""" + + whitelist_path = tmp_path / "whitelist.csv" + whitelist_path.write_text("10\n", encoding="utf-8") + monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", VALID_BOT_TOKEN) + monkeypatch.setenv("TELEGRAM_RESENDER_FORWARD_CHAT_ID", "100") + monkeypatch.setenv("TELEGRAM_RESENDER_WHITELIST_PATH", str(whitelist_path)) + monkeypatch.setenv("TELEGRAM_RESENDER_STORAGE_PATH", str(tmp_path)) + stdout = StringIO() + stderr = StringIO() + + assert cli.run_health(stdout=stdout, stderr=stderr) == 2 + + assert stdout.getvalue() == "" + assert "health=error" in stderr.getvalue() + + +def test_health_rejects_incompatible_sqlite_schema( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Health must validate the delivery table rather than only its name.""" + + whitelist_path = tmp_path / "whitelist.csv" + whitelist_path.write_text("10\n", encoding="utf-8") + storage_path = tmp_path / "requests.sqlite3" + connection = sqlite3.connect(storage_path) + try: + connection.execute("CREATE TABLE request_deliveries (request_id TEXT)") + connection.commit() + finally: + connection.close() + monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", VALID_BOT_TOKEN) + monkeypatch.setenv("TELEGRAM_RESENDER_FORWARD_CHAT_ID", "100") + monkeypatch.setenv("TELEGRAM_RESENDER_WHITELIST_PATH", str(whitelist_path)) + monkeypatch.setenv("TELEGRAM_RESENDER_STORAGE_PATH", str(storage_path)) + stdout = StringIO() + stderr = StringIO() + + assert cli.run_health(stdout=stdout, stderr=stderr) == 2 + + assert stdout.getvalue() == "" + assert "schema mismatch" in stderr.getvalue() + + +def test_run_bot_formats_sqlite_startup_errors( + monkeypatch: pytest.MonkeyPatch, + capsys: Any, +) -> None: + """SQLite startup failures should use the same concise CLI error path.""" + + async def fail_storage() -> None: + raise sqlite3.OperationalError("unable to open database file") + + monkeypatch.setattr(cli, "run_polling", fail_storage) + + assert cli.run_bot() == 2 + captured = capsys.readouterr() + assert "Startup error: unable to open database file" in captured.err + assert "Traceback" not in captured.err diff --git a/tests/unit/test_delivery.py b/tests/unit/test_delivery.py index 114643d..1bfbbde 100644 --- a/tests/unit/test_delivery.py +++ b/tests/unit/test_delivery.py @@ -13,6 +13,7 @@ async def test_send_with_retry_retries_telegram_api_errors() -> None: attempts = 0 sleeps: list[float] = [] + lease_extensions: list[float] = [] async def fake_send(chat_id: int, text: str) -> None: nonlocal attempts @@ -33,10 +34,12 @@ async def fake_sleep(delay: float) -> None: max_attempts=2, backoff_seconds=0.5, sleep=fake_sleep, + before_retry_wait=lease_extensions.append, ) assert attempts == 2 assert sleeps == [0.5] + assert lease_extensions == [0.5] @pytest.mark.asyncio diff --git a/tests/unit/test_formatting.py b/tests/unit/test_formatting.py index be4f982..a648ec3 100644 --- a/tests/unit/test_formatting.py +++ b/tests/unit/test_formatting.py @@ -1,5 +1,6 @@ """Unit tests for message rendering rules.""" +import hashlib from datetime import UTC, datetime from telegram_resender.formatting import MessageFormatter @@ -26,3 +27,32 @@ def test_message_formatter_includes_source_and_author() -> None: assert "From: Alex M (@TestUser)" in rendered assert "Source chat: 1001" in rendered assert "Tower A" in rendered + + +def test_fallback_request_id_uses_immutable_user_id() -> None: + """Fallback IDs must not collide across users or change with mutable usernames.""" + + formatter = MessageFormatter() + first = IncomingMessage( + chat_id=-100, + text="same request", + user=UserProfile(id=10, username="old-name"), + ) + renamed = IncomingMessage( + chat_id=-100, + text="same request", + user=UserProfile(id=10, username="new-name"), + ) + other_user = IncomingMessage( + chat_id=-100, + text="same request", + user=UserProfile(id=20, username="old-name"), + ) + + assert formatter.format_request_id(first) == formatter.format_request_id(renamed) + assert formatter.format_request_id(first) != formatter.format_request_id(other_user) + + legacy_digest = hashlib.sha256( + f"{first.chat_id}|{first.user.username or ''}|{first.text}".encode() + ).hexdigest() + assert formatter.format_legacy_request_id(first) == f"local-{legacy_digest[:12]}" diff --git a/tests/unit/test_routes.py b/tests/unit/test_routes.py index bbb971b..2cd58ff 100644 --- a/tests/unit/test_routes.py +++ b/tests/unit/test_routes.py @@ -1,5 +1,6 @@ """Unit tests for route config parsing and matching.""" +import json from pathlib import Path import pytest @@ -18,6 +19,7 @@ def test_load_routes_parses_json_config(tmp_path: Path) -> None: { "name": "gate", "target_chat_id": 100, + "allowed_user_ids": [10], "allowed_usernames": ["@Alice"], "keywords_any": ["tower"], "keywords_none": ["cancel"], @@ -32,9 +34,10 @@ def test_load_routes_parses_json_config(tmp_path: Path) -> None: routes = load_routes(path) assert len(routes) == 1 - assert routes[0].matches(username="alice", text="Tower A request") is True - assert routes[0].matches(username="bob", text="Tower A request") is False - assert routes[0].matches(username="alice", text="Tower A cancel") is False + assert routes[0].matches(user_id=10, username="alice", text="Tower A request") is True + assert routes[0].matches(user_id=20, username="alice", text="Tower A request") is False + assert routes[0].matches(user_id=10, username="bob", text="Tower A request") is False + assert routes[0].matches(user_id=10, username="alice", text="Tower A cancel") is False assert routes[0].render_forward_text("payload") == "[gate]\npayload" @@ -46,3 +49,163 @@ def test_load_routes_rejects_empty_config(tmp_path: Path) -> None: with pytest.raises(ValueError, match="at least one route"): load_routes(path) + + +def test_load_routes_rejects_invalid_template_at_startup(tmp_path: Path) -> None: + """Unknown template fields should fail validation before handling user messages.""" + + path = tmp_path / "routes.json" + path.write_text( + '{"routes": [{"name": "gate", "target_chat_id": 100, "template": "{missing}"}]}', + encoding="utf-8", + ) + + with pytest.raises(ValueError, match="invalid template"): + load_routes(path) + + +@pytest.mark.parametrize("template", [" ", "x" * 4097]) +def test_load_routes_rejects_unsendable_template(tmp_path: Path, template: str) -> None: + """A route template that cannot produce a Telegram message should fail at startup.""" + + path = tmp_path / "routes.json" + path.write_text( + json.dumps({"routes": [{"name": "gate", "target_chat_id": 100, "template": template}]}), + encoding="utf-8", + ) + + with pytest.raises(ValueError, match="template"): + load_routes(path) + + +@pytest.mark.parametrize( + ("route_fragment", "error"), + [ + ('"enabled": "false"', "enabled must be a boolean"), + ('"target_chat_id": 1.5', "target_chat_id must be a non-zero integer"), + ('"keywords_any": 42', "keyword filters must be"), + ], +) +def test_load_routes_rejects_mistyped_fields( + tmp_path: Path, + route_fragment: str, + error: str, +) -> None: + """Mistyped JSON values must not be silently coerced into unsafe routes.""" + + path = tmp_path / "routes.json" + base = '"name": "gate", "target_chat_id": 100' + if route_fragment.startswith('"target_chat_id"'): + base = '"name": "gate"' + path.write_text(f'{{"routes": [{{{base}, {route_fragment}}}]}}', encoding="utf-8") + + with pytest.raises(ValueError, match=error): + load_routes(path) + + +@pytest.mark.parametrize( + "unknown_field", + ["allowed_username", "enable", "keyword_any"], +) +def test_load_routes_rejects_unknown_route_fields_fail_closed( + tmp_path: Path, + unknown_field: str, +) -> None: + """A route typo must not silently remove an access or keyword restriction.""" + + path = tmp_path / "routes.json" + path.write_text( + f'{{"routes": [{{"name": "gate", "target_chat_id": 100, "{unknown_field}": []}}]}}', + encoding="utf-8", + ) + + with pytest.raises(ValueError, match=rf"unknown field.*{unknown_field}"): + load_routes(path) + + +def test_load_routes_rejects_unknown_document_fields(tmp_path: Path) -> None: + """The wrapper object should use a closed schema as well.""" + + path = tmp_path / "routes.json" + path.write_text('{"route": []}', encoding="utf-8") + + with pytest.raises(ValueError, match=r"unknown field.*'route'"): + load_routes(path) + + +@pytest.mark.parametrize("field_name", ["allowed_usernames", "keywords_any", "keywords_none"]) +def test_load_routes_rejects_blank_filter_values(tmp_path: Path, field_name: str) -> None: + """Blank filters must not normalize into an unrestricted route.""" + + path = tmp_path / "routes.json" + path.write_text( + f'{{"routes": [{{"name": "gate", "target_chat_id": 100, "{field_name}": [""]}}]}}', + encoding="utf-8", + ) + + with pytest.raises(ValueError, match="non-empty strings"): + load_routes(path) + + +def test_load_routes_rejects_duplicate_json_keys(tmp_path: Path) -> None: + """Duplicate keys must not redirect a route or erase an authorization filter.""" + + path = tmp_path / "routes.json" + path.write_text( + """ + { + "routes": [{ + "name": "private", + "target_chat_id": -100, + "target_chat_id": -999, + "allowed_user_ids": [10], + "allowed_user_ids": [] + }] + } + """, + encoding="utf-8", + ) + + with pytest.raises(ValueError, match=r"duplicate JSON key 'target_chat_id'"): + load_routes(path) + + +def test_load_routes_rejects_username_that_normalizes_empty(tmp_path: Path) -> None: + """A placeholder '@' must not silently turn a restricted route into an open route.""" + + path = tmp_path / "routes.json" + path.write_text( + '{"routes": [{"target_chat_id": 100, "allowed_usernames": [" @ "]}]}', + encoding="utf-8", + ) + + with pytest.raises(ValueError, match="valid non-empty usernames"): + load_routes(path) + + +@pytest.mark.parametrize("value", [0, -1, True, "not-an-id", 1.5]) +def test_load_routes_rejects_invalid_allowed_user_ids(tmp_path: Path, value: object) -> None: + """Numeric route authorization must fail closed for malformed IDs.""" + + path = tmp_path / "routes.json" + path.write_text( + json.dumps({"routes": [{"target_chat_id": 100, "allowed_user_ids": [value]}]}), + encoding="utf-8", + ) + + with pytest.raises(ValueError, match="allowed_user_ids"): + load_routes(path) + + +def test_legacy_username_only_route_remains_compatible(tmp_path: Path) -> None: + """Existing routes keep their legacy username routing behavior during migration.""" + + path = tmp_path / "routes.json" + path.write_text( + '{"routes": [{"target_chat_id": 100, "allowed_usernames": ["alice"]}]}', + encoding="utf-8", + ) + + route = load_routes(path)[0] + assert route.allowed_user_ids == frozenset() + assert route.matches(user_id=999, username="alice", text="request") is True diff --git a/tests/unit/test_service.py b/tests/unit/test_service.py index 87bd3f5..870d1d3 100644 --- a/tests/unit/test_service.py +++ b/tests/unit/test_service.py @@ -109,6 +109,13 @@ def test_service_authorizes_by_user_id_not_username() -> None: assert reclaimed_username.reason == "unknown_user_id" +def test_access_denied_guidance_references_numeric_user_id() -> None: + """Denial guidance must describe the numeric-ID whitelist actually in use.""" + + assert "user ID" in RU_MESSAGES.access_denied_unknown + assert "/whoami" in RU_MESSAGES.access_denied_unknown + + def test_service_reports_missing_template_fields() -> None: """A labeled but incomplete request should name the missing fields.""" @@ -181,6 +188,129 @@ def test_service_routes_request_to_matching_destinations() -> None: assert service.render_forward_text_for_target(300, decision.forward_text).startswith("[all]") +def test_service_uses_template_from_the_route_that_actually_matched() -> None: + """Shared target chats must not pick a template from a non-matching route.""" + + service = ResenderService( + whitelist=Whitelist([10]), + formatter=MessageFormatter(), + request_accepted_message="ok", + access_denied_message="deny", + missing_username_message=RU_MESSAGES.access_denied_missing_username, + invalid_request_message=RU_MESSAGES.invalid_request, + missing_fields_message=RU_MESSAGES.missing_fields, + no_route_matched_message=RU_MESSAGES.no_route_matched, + locale="ru", + routes=( + RouteRule( + name="bob-route", + target_chat_id=200, + allowed_usernames=frozenset({"bob"}), + keywords_any=(), + keywords_none=(), + template="[bob]\n{request}", + ), + RouteRule( + name="alice-route", + target_chat_id=200, + allowed_usernames=frozenset({"alice"}), + keywords_any=(), + keywords_none=(), + template="[alice]\n{request}", + ), + ), + ) + + decision = service.handle_text( + IncomingMessage(chat_id=1, text=VALID_REQUEST, user=UserProfile(id=10, username="alice")) + ) + + assert decision.should_forward is True + assert len(decision.forward_payloads) == 1 + assert decision.forward_payloads[0][0] == 200 + assert decision.forward_payloads[0][1].startswith("[alice]") + + +def test_numeric_route_filter_cannot_be_bypassed_by_changing_username() -> None: + """A globally allowed user must not enter another numeric-ID route by renaming.""" + + service = ResenderService( + whitelist=Whitelist([10, 20]), + formatter=MessageFormatter(), + request_accepted_message="ok", + access_denied_message="deny", + missing_username_message=RU_MESSAGES.access_denied_missing_username, + invalid_request_message=RU_MESSAGES.invalid_request, + missing_fields_message=RU_MESSAGES.missing_fields, + no_route_matched_message=RU_MESSAGES.no_route_matched, + locale="ru", + routes=( + RouteRule( + name="private", + target_chat_id=200, + allowed_usernames=frozenset(), + keywords_any=(), + keywords_none=(), + template=None, + allowed_user_ids=frozenset({10}), + ), + ), + ) + + allowed = service.handle_text( + IncomingMessage(chat_id=1, text=VALID_REQUEST, user=UserProfile(id=10, username="new")) + ) + renamed = service.handle_text( + IncomingMessage(chat_id=1, text=VALID_REQUEST, user=UserProfile(id=20, username="new")) + ) + + assert allowed.should_forward is True + assert renamed.should_forward is False + assert renamed.reason == "no_route_matched" + + +def test_first_matching_route_wins_for_shared_target() -> None: + """One target receives one deterministic payload even when multiple rules match.""" + + service = ResenderService( + whitelist=Whitelist([10]), + formatter=MessageFormatter(), + request_accepted_message="ok", + access_denied_message="deny", + missing_username_message=RU_MESSAGES.access_denied_missing_username, + invalid_request_message=RU_MESSAGES.invalid_request, + missing_fields_message=RU_MESSAGES.missing_fields, + no_route_matched_message=RU_MESSAGES.no_route_matched, + locale="ru", + routes=( + RouteRule( + name="first", + target_chat_id=200, + allowed_usernames=frozenset(), + keywords_any=(), + keywords_none=(), + template="[first]\n{request}", + ), + RouteRule( + name="second", + target_chat_id=200, + allowed_usernames=frozenset(), + keywords_any=(), + keywords_none=(), + template="[second]\n{request}", + ), + ), + ) + + decision = service.handle_text( + IncomingMessage(chat_id=1, text=VALID_REQUEST, user=UserProfile(id=10, username="alice")) + ) + + assert decision.target_chat_ids == (200,) + assert len(decision.forward_payloads) == 1 + assert decision.forward_payloads[0][1].startswith("[first]") + + def test_service_matches_route_keywords_only_against_building_field() -> None: """Keywords in free-form fields must not add unintended destinations.""" diff --git a/tests/unit/test_storage.py b/tests/unit/test_storage.py index 6142793..682faa0 100644 --- a/tests/unit/test_storage.py +++ b/tests/unit/test_storage.py @@ -1,10 +1,19 @@ """Unit tests for SQLite request delivery storage.""" +import sqlite3 from datetime import date from io import StringIO from pathlib import Path -from telegram_resender.storage import RequestLog, export_records_csv +import pytest + +import telegram_resender.storage as storage_module +from telegram_resender.storage import ( + DeliveryInProgressError, + PendingRequestStore, + RequestLog, + export_records_csv, +) def test_request_log_tracks_delivery_idempotently(tmp_path: Path) -> None: @@ -12,12 +21,13 @@ def test_request_log_tracks_delivery_idempotently(tmp_path: Path) -> None: request_log = RequestLog(tmp_path / "requests.sqlite3") - assert request_log.begin_delivery( + lease = request_log.begin_delivery( request_id="req-1", target_chat_id=100, sender_username="alice", ) - request_log.mark_delivery(request_id="req-1", target_chat_id=100, status="delivered") + assert lease is not None + request_log.mark_delivery(lease=lease, status="delivered") assert not request_log.begin_delivery( request_id="req-1", @@ -26,16 +36,344 @@ def test_request_log_tracks_delivery_idempotently(tmp_path: Path) -> None: ) +def test_request_log_honors_legacy_request_id_alias(tmp_path: Path) -> None: + """A delivered pre-upgrade local ID must suppress its new immutable-ID replacement.""" + + request_log = RequestLog(tmp_path / "requests.sqlite3") + lease = request_log.begin_delivery( + request_id="local-legacy", + target_chat_id=100, + sender_username="alice", + ) + assert lease is not None + request_log.mark_delivery(lease=lease, status="delivered") + + assert not request_log.begin_delivery( + request_id="local-current", + request_id_aliases=("local-legacy",), + target_chat_id=100, + sender_username="alice", + ) + + +def test_delivery_lease_blocks_concurrency_and_rejects_stale_owner( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Only one owner may deliver, and an expired owner cannot finish a newer lease.""" + + now = [100.0] + monkeypatch.setattr(storage_module, "time", lambda: now[0]) + request_log = RequestLog(tmp_path / "requests.sqlite3", lease_seconds=30) + + first = request_log.begin_delivery( + request_id="req-1", + target_chat_id=100, + sender_username="alice", + ) + assert first is not None + with pytest.raises(DeliveryInProgressError): + request_log.begin_delivery( + request_id="req-1", + target_chat_id=100, + sender_username="alice", + ) + + now[0] += 20 + assert request_log.renew_delivery(first, wait_seconds=20) is True + now[0] += 15 + with pytest.raises(DeliveryInProgressError): + request_log.begin_delivery( + request_id="req-1", + target_chat_id=100, + sender_username="alice", + ) + + now[0] += 36 + second = request_log.begin_delivery( + request_id="req-1", + target_chat_id=100, + sender_username="alice", + ) + assert second is not None + assert second.version == first.version + 1 + assert second.owner_token != first.owner_token + assert request_log.mark_delivery(lease=first, status="delivered") is False + assert request_log.mark_delivery(lease=second, status="delivered") is True + assert ( + request_log.begin_delivery( + request_id="req-1", + target_chat_id=100, + sender_username="alice", + ) + is None + ) + + +def test_delivery_lease_alias_rejects_partially_superseded_owner( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reclaiming a legacy alias must fence the old owner of the full alias set.""" + + now = [100.0] + monkeypatch.setattr(storage_module, "time", lambda: now[0]) + request_log = RequestLog(tmp_path / "requests.sqlite3", lease_seconds=30) + first = request_log.begin_delivery( + request_id="local-current", + request_id_aliases=("local-legacy",), + target_chat_id=100, + sender_username="alice", + ) + assert first is not None + + now[0] += 31 + second = request_log.begin_delivery( + request_id="local-legacy", + target_chat_id=100, + sender_username="alice", + ) + assert second is not None + + assert request_log.renew_delivery(first) is False + assert request_log.mark_delivery(lease=first, status="delivered") is False + assert request_log.mark_delivery(lease=second, status="delivered") is True + assert ( + request_log.begin_delivery( + request_id="local-current", + request_id_aliases=("local-legacy",), + target_chat_id=100, + sender_username="alice", + ) + is None + ) + + +def test_pending_store_persists_and_versions_stale_claims( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Pending ownership must survive restart and reject a superseded worker.""" + + now = [100.0] + monkeypatch.setattr(storage_module, "time", lambda: now[0]) + path = tmp_path / "requests.sqlite3" + first_store = PendingRequestStore(path, ttl_seconds=120, claim_seconds=30) + published = first_store.publish( + (100, 10), + request_id="req-1", + request_id_aliases=("legacy-1",), + forward_text_by_chat_id=((200, "payload"),), + sender_user_id=10, + sender_username="alice", + ) + + restarted_store = PendingRequestStore(path, ttl_seconds=120, claim_seconds=30) + first_claim = restarted_store.claim((100, 10), request_id="req-1") + assert first_claim is not None + assert first_claim.version == published.version + assert first_store.claim((100, 10), request_id="req-1") is None + + now[0] += 31 + second_claim = first_store.claim((100, 10), request_id="req-1") + assert second_claim is not None + assert second_claim.version == first_claim.version + assert second_claim.owner_version == first_claim.owner_version + 1 + assert second_claim.owner_token != first_claim.owner_token + assert restarted_store.release((100, 10), first_claim) is False + assert restarted_store.complete((100, 10), first_claim) is False + assert first_store.release((100, 10), second_claim) is True + + third_claim = restarted_store.claim((100, 10), request_id="req-1") + assert third_claim is not None + assert third_claim.owner_version == second_claim.owner_version + 1 + assert restarted_store.complete((100, 10), third_claim) is True + assert first_store.claim((100, 10), request_id="req-1") is None + + +def test_pending_publish_replaces_only_unclaimed_duplicate( + tmp_path: Path, +) -> None: + """Replayed previews with one ID must not leave hidden duplicate confirmations.""" + + store = PendingRequestStore(tmp_path / "requests.sqlite3", ttl_seconds=120) + first = store.publish( + (100, 10), + request_id="req-1", + request_id_aliases=(), + forward_text_by_chat_id=((200, "old"),), + sender_user_id=10, + sender_username="alice", + ) + second = store.publish( + (100, 10), + request_id="req-1", + request_id_aliases=(), + forward_text_by_chat_id=((200, "new"),), + sender_user_id=10, + sender_username="alice", + ) + assert second.version > first.version + + cancelled = store.cancel((100, 10), request_id="req-1") + assert cancelled is not None + assert cancelled.forward_text_by_chat_id == ((200, "new"),) + assert store.cancel((100, 10), request_id="req-1") is None + + +def test_pending_claim_renewal_covers_retry_wait( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A known retry delay must not let another confirmation supersede its owner.""" + + now = [100.0] + monkeypatch.setattr(storage_module, "time", lambda: now[0]) + store = PendingRequestStore(tmp_path / "requests.sqlite3", ttl_seconds=120, claim_seconds=30) + store.publish( + (100, 10), + request_id="req-1", + request_id_aliases=(), + forward_text_by_chat_id=((200, "payload"),), + sender_user_id=10, + sender_username="alice", + ) + first = store.claim((100, 10), request_id="req-1") + assert first is not None + + now[0] += 20 + assert store.renew((100, 10), first, wait_seconds=20) is True + now[0] += 20 + assert store.claim((100, 10), request_id="req-1") is None + + now[0] += 31 + second = store.claim((100, 10), request_id="req-1") + assert second is not None + assert second.owner_version == first.owner_version + 1 + + +def test_releasing_old_claim_keeps_only_newest_duplicate_preview(tmp_path: Path) -> None: + """A replay published during delivery must replace the old ID after failure.""" + + store = PendingRequestStore(tmp_path / "requests.sqlite3", ttl_seconds=120) + store.publish( + (100, 10), + request_id="req-1", + request_id_aliases=(), + forward_text_by_chat_id=((200, "old"),), + sender_user_id=10, + sender_username="alice", + ) + claimed_old = store.claim((100, 10), request_id="req-1") + assert claimed_old is not None + store.publish( + (100, 10), + request_id="req-1", + request_id_aliases=(), + forward_text_by_chat_id=((200, "new"),), + sender_user_id=10, + sender_username="alice", + ) + + assert store.release((100, 10), claimed_old) is True + newest = store.cancel((100, 10), request_id="req-1") + assert newest is not None + assert newest.forward_text_by_chat_id == ((200, "new"),) + assert store.cancel((100, 10), request_id="req-1") is None + + +def test_request_log_rejects_schema_without_unique_delivery_index(tmp_path: Path) -> None: + """The schema check must verify the uniqueness required by ON CONFLICT/idempotency.""" + + storage_path = tmp_path / "requests.sqlite3" + connection = sqlite3.connect(storage_path) + try: + connection.execute( + """ + CREATE TABLE request_deliveries ( + request_id TEXT NOT NULL, + target_chat_id INTEGER NOT NULL, + sender_username TEXT, + created_at TEXT NOT NULL, + validation_status TEXT NOT NULL, + delivery_status TEXT NOT NULL, + last_error TEXT + ) + """ + ) + connection.commit() + finally: + connection.close() + + with pytest.raises(sqlite3.DatabaseError, match="missing primary key"): + RequestLog(storage_path) + + +def test_existing_delivery_database_gains_new_state_tables_without_data_loss( + tmp_path: Path, +) -> None: + """Upgrading a valid pre-lease database must preserve delivered rows.""" + + storage_path = tmp_path / "requests.sqlite3" + connection = sqlite3.connect(storage_path) + try: + connection.execute( + """ + CREATE TABLE request_deliveries ( + request_id TEXT NOT NULL, + target_chat_id INTEGER NOT NULL, + sender_username TEXT, + created_at TEXT NOT NULL, + validation_status TEXT NOT NULL, + delivery_status TEXT NOT NULL, + last_error TEXT, + PRIMARY KEY (request_id, target_chat_id) + ) + """ + ) + connection.execute( + """ + INSERT INTO request_deliveries VALUES + ('old-delivered', 100, 'alice', '2026-01-01T00:00:00+00:00', + 'valid', 'delivered', NULL) + """ + ) + connection.commit() + finally: + connection.close() + + request_log = RequestLog(storage_path) + assert ( + request_log.begin_delivery( + request_id="old-delivered", + target_chat_id=100, + sender_username="alice", + ) + is None + ) + pending_store = PendingRequestStore(storage_path, ttl_seconds=120) + published = pending_store.publish( + (100, 10), + request_id="new-pending", + request_id_aliases=(), + forward_text_by_chat_id=((200, "payload"),), + sender_user_id=10, + sender_username="alice", + ) + assert published.version == 1 + + def test_request_log_exports_csv(tmp_path: Path) -> None: """Request log records should export as stable CSV.""" request_log = RequestLog(tmp_path / "requests.sqlite3") - request_log.begin_delivery( + lease = request_log.begin_delivery( request_id="req-1", target_chat_id=100, sender_username="alice", ) - request_log.mark_delivery(request_id="req-1", target_chat_id=100, status="failed", error="boom") + assert lease is not None + request_log.mark_delivery(lease=lease, status="failed", error="boom") records = request_log.records_since(since=date(2000, 1, 1)) output = StringIO() @@ -50,17 +388,13 @@ def test_request_log_export_neutralizes_spreadsheet_formulas(tmp_path: Path) -> """CSV export should not emit formula-prefixed text fields directly.""" request_log = RequestLog(tmp_path / "requests.sqlite3") - request_log.begin_delivery( + lease = request_log.begin_delivery( request_id="=req-1", target_chat_id=100, sender_username="@alice", ) - request_log.mark_delivery( - request_id="=req-1", - target_chat_id=100, - status="failed", - error="+boom", - ) + assert lease is not None + request_log.mark_delivery(lease=lease, status="failed", error="+boom") records = request_log.records_since(since=date(2000, 1, 1)) output = StringIO() diff --git a/tests/unit/test_telegram_limits.py b/tests/unit/test_telegram_limits.py new file mode 100644 index 0000000..d4953ea --- /dev/null +++ b/tests/unit/test_telegram_limits.py @@ -0,0 +1,20 @@ +"""Unit tests for Telegram UTF-16 message limits.""" + +from telegram_resender.telegram_limits import fits_telegram_message, telegram_utf16_length + + +def test_telegram_message_limit_uses_utf16_code_units() -> None: + """Astral characters count as two units and exact boundary values remain valid.""" + + assert telegram_utf16_length("😀") == 2 + assert fits_telegram_message("x" * 4096) is True + assert fits_telegram_message("x" * 4097) is False + assert fits_telegram_message("😀" * 2048) is True + assert fits_telegram_message("😀" * 2049) is False + + +def test_telegram_message_limit_rejects_empty_or_invalid_unicode() -> None: + """sendMessage requires non-empty valid Unicode text.""" + + assert fits_telegram_message("") is False + assert fits_telegram_message("\ud800") is False diff --git a/tests/unit/test_whitelist.py b/tests/unit/test_whitelist.py index 728c5b8..b1767e3 100644 --- a/tests/unit/test_whitelist.py +++ b/tests/unit/test_whitelist.py @@ -42,3 +42,11 @@ def test_parse_user_id_rejects_usernames() -> None: assert parse_user_id(" 123 ") == 123 with pytest.raises(ValueError): parse_user_id("alice") + + +@pytest.mark.parametrize("value", [0, -1, True, "-100"]) +def test_parse_user_id_rejects_non_positive_values(value: int | str) -> None: + """Telegram sender user IDs are positive integers, never chat IDs or booleans.""" + + with pytest.raises(ValueError, match="positive integer"): + parse_user_id(value)