10 Commits

Author SHA1 Message Date
Никита Sonic 8617d25f3a chore(release): 2.0.0-beta.14 (#57)
Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-27 02:40:24 +03:00
Никита Sonic e9af114f3c fix(worker): запись в Telegram шла без ожидания и backpressure (#42) (#56)
Скрипт воркера писал в сокет Telegram так:

    server.addEventListener("message", (event) => {
      writer.write(chunk).catch(shutdown);
    });

`write()` вызывался поверх незавершённого, `writer.ready` не спрашивался вовсе.
Пока в клиенте голодала отправка, настоящего потока вверх через воркер не
возникало, и код держался. В beta.12 голодание починили — поток появился, и
репортёр #42 сразу получил переподключения на обоих клиентах, которых на
beta.11 с тем же воркером не было.

Запись сериализована цепочкой промисов: следующий чанк уходит после того, как
записан предыдущий, и только когда писатель готов. Кто разворачивал воркер
раньше — нужен передеплой, о чём сказано в docs/CLOUDFLARE_WORKER.md.

Причина у репортёра не подтверждена: рантайма Workers у меня нет, проверить
можно только у него.

Заодно счётчик «промолчали». Соединение, которое открылось и ничего не
прислало за `IO_TIMEOUT`, закрывалось и не попадало ни в один счётчик:
`unknown_clients` растёт, только когда запрос пришёл и не разобрался, а не
когда его не дождались. Тот же репортёр сообщил, что его телефон
переустанавливает соединение примерно раз в десять секунд — ровно период
`IO_TIMEOUT`. Проверить это по диагностике было нечем, теперь есть чем.

Тесты: Ping через туннель (путь не был покрыт вовсе, а Ping бывает только на
маршруте воркера) и молчащий клиент. Второй гоняет виртуальное время, чтобы не
ждать десять секунд по-настоящему, — отсюда dev-зависимость на tokio/test-util.

Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-27 02:34:20 +03:00
Никита Sonic 935121b991 chore(release): 2.0.0-beta.13 (#55)
Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-26 17:05:46 +03:00
Никита Sonic fe9aad5ee9 fix(diag): причина отказа туннеля и судьба домена Worker'а попадали в никуда (#50) (#54)
`TransportEngine::connect` собирает подробный перечень попыток — какой адрес не
ответил, где истёк TLS, что вернул воркер, — и возвращает его в `Err`. Дальше
этот `Err` доходил до `serve`, где выбрасывался: `let _ = handle(...)`.
Увеличивался только счётчик.

Снаружи это выглядит как `туннелей 0 · сбоев 249 · падений маршрутов 395` без
единого слова о том, почему их ноль. Отличить «провайдер режет закреплённые
адреса» от «воркер отвечает отказом» нечем, хотя рядом есть журнал событий, в
который пишутся куда менее важные вещи.

Теперь причина попадает в журнал строкой вида:

    Не поднялся туннель до DC2: 149.154.167.51 — не отвечает (таймаут TCP);
    kws2.web.telegram.org — таймаут TLS/WebSocket

Дедупликация журнала делает её разовой: набор маршрутов у DC стабилен.

Там же вторая слепая зона. Домен воркера, не похожий на имя хоста, отбрасывался
молча: `https://name.workers.dev/` со схемой или слэшем не проходит
`valid_domain`, маршрут не появляется, и «воркер настроен» неотличимо от
«воркера нет». `set_worker_domains` теперь возвращает принятые и отвергнутые
по отдельности, отвергнутые называются вместе с причиной, принятые
подтверждаются.

Тексты отказов переведены на русский: их читает не разработчик, а человек,
который прислал скриншот и ждёт ответа.

Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-26 16:59:13 +03:00
Никита Sonic 28085f9127 chore(release): 2.0.0-beta.12 (#52)
Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-26 16:34:54 +03:00
Никита Sonic 923c22f9b4 fix(proxy): загрузка останавливала отправку, туннель шёл в одну сторону (#42, #32) (#51)
Оба направления туннеля обслуживал один `select!` с пометкой `biased`.
`biased` опрашивает ветки строго по порядку: пока в первой — «Telegram →
клиент» — есть данные, до второй очередь не доходит вообще. При непрерывном
потоке вниз, то есть при первичной синхронизации телефона или загрузке медиа,
исходящие пакеты клиента не читались.

Второй дефект в том же цикле: одна задача на оба направления. `tcp_w.write_all`
ждёт, пока клиент разберёт присланное, и всё это время не опрашивается чтение
от клиента. Телефон по Wi-Fi разбирает поток медленнее, чем Telegram Desktop на
той же машине через loopback, — отсюда асимметрия «на компьютере работает, на
телефоне нет».

Для MTProto это фатально: клиент обязан слать подтверждения, а за каждым
следующим куском файла — свой `upload.getFile`. Первый запрос уходит, дальше
идёт поток вниз, и следующие запросы наверх не попадают. Снаружи это выглядит
как «Подключено» при живом туннеле, нулевых сбоях и нулевых отклонениях: чаты
на месте, иконки не грузятся, отправка виснет с часиками.

Направления разделены на две независимые половины: `ws.split()` плюс
`CryptoContext::split()`, потому что шифры направлений независимы — два потока
AES-CTR со своими ключами. Ping приходит в читающую половину, а отвечает на
него пишущая, через канал на четыре слота: владелец отправляющей половины
должен оставаться ровно один.

Два теста, падающие на beta.11: за пять секунд непрерывной загрузки наверх не
уходит ни одного байта, и клиент, не успевающий читать, замораживает
собственную отправку. Снятие одного `biased` чинит только первый — это и
показывает, что дефекта два.

Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-26 16:28:43 +03:00
Никита Sonic 8ce3c368a3 chore(release): 2.0.0-beta.11 (#48)
Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-19 15:04:34 +03:00
Никита Sonic 0dc6b7bf3e fix(proxy): DC и маршрут в строке статуса были из разных соединений (#47)
В диагностике из #42 встречаются строки вида

    соединений 9 · туннелей 9 · DC5 · Запасной Telegram IP · сбоев 12

Такого сочетания не бывает: у DC1, DC3, DC5 и DC203 закреплённый адрес ровно
один, и маршрута «запасной адрес» у них не существует в принципе. Значит номер
и маршрут пришли из разных соединений.

Так и было. `last_dc` писало соединение при разборе init, `last_route` — другое
соединение после рукопожатия, двумя независимыми атомиками. У репортёра от
пяти до двадцати шести одновременных соединений, поэтому пара складывалась
случайно. Читается она как «до этого DC шли этим маршрутом» и в этом качестве
врала — ровно тот класс дефектов, ради которого затевалась честная диагностика
в #38.

Теперь пара пишется одним значением в момент, когда туннель поднялся:
`dc << 8 | route`. Пока туннеля не было, показывается разобранный DC и
«маршрут ещё не выбран» — это состояние тоже настоящее и его терять не надо.

Поля стали приватными, наружу выведены `last_dc()` и `last_route()`, чтобы
рассогласовать их снаружи было нельзя.

Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-19 14:59:34 +03:00
Никита Sonic fe9ab39abe chore(release): 2.0.0-beta.10 (#46)
Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-19 14:37:05 +03:00
Никита Sonic b92403b4c9 fix(proxy): клиент, с которым не договорились, закрывался молча (#42) (#45)
Диагностика от @alexsagaidak в #42 показала третий случай, которого ни один
счётчик не различал. У него ноль отклонённых и живые туннели, то есть оба
показателя говорят «всё хорошо»:

    соединений 17 · туннелей 6 · DC4 · Запасной Telegram IP · сбоев 0 ·
    падений маршрутов 26 · отклонено 0

Клиент, который дошёл до прокси, но не сумел договориться, не попадал ни в
blocked, ни в ws_failures. Ошибка из handle() выбрасывалась в `let _ =`,
соединение закрывалось, и наружу это выглядело как active, дёрнувшийся вверх и
обратно. По диагностике неотличимо от клиента, который подключился и работает.

Теперь такие клиенты считает unknown_clients, а журнал называет адрес и
причину. Причин две:

MTProto-init не разбирается под текущим секретом. Почти всегда это ссылка
tg://proxy от прошлого запуска: секрет — её половина, и клиент с сохранённой
старой ссылкой попадает ровно сюда. Со стороны Telegram это и есть «прокси
настроен неверно и будет отключён» — то, с чем пришли в #37 и что до сих пор
нельзя было подтвердить со стороны прокси.

SOCKS5-приветствие не разбирается. Сюда же попадает MTProto-соединение,
ушедшее в SOCKS5-ветку по неоднозначному первому байту, если полный init не
успел прийти за PROTOCOL_PROBE_TIMEOUT. На loopback этого не бывает, а через
Wi-Fi с телефона — уже вопрос задержки.

От ws_failures отличается тем, что тот растёт после успешного рукопожатия с
клиентом: там договорились с клиентом, но не смогли с Telegram. Различать их
важно, иначе непонятно, в какую сторону смотреть.

В интерфейсе — метрика «Не опознаны» с пояснением, в строке статуса CLI —
поле «не опознано N».

Co-authored-by: by-sonic <171230345+by-sonic@users.noreply.github.com>
2026-08-19 14:32:34 +03:00
14 changed files with 964 additions and 92 deletions
Generated
+1 -1
View File
@@ -3553,7 +3553,7 @@ dependencies = [
[[package]]
name = "tglock"
version = "2.0.0-beta.9"
version = "2.0.0-beta.14"
dependencies = [
"aes",
"cipher",
+7 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "tglock"
version = "2.0.0-beta.9"
version = "2.0.0-beta.14"
edition = "2021"
rust-version = "1.88"
description = "Telegram unblock via local WebSocket tunnel"
@@ -65,3 +65,9 @@ rand = "0.8"
[build-dependencies]
tauri-build = { version = "2", features = [], optional = true }
[dev-dependencies]
# `start_paused` в тестах: таймаут ожидания запроса от клиента — десять секунд,
# и ждать их по-настоящему в тесте нельзя. В сборку не попадает: dev-зависимости
# участвуют только в тестах.
tokio = { version = "1", features = ["test-util"] }
+2 -1
View File
@@ -169,6 +169,7 @@ Telegram → Настройки → **Продвинутые** → Тип сое
>
> - **Соединения `0` и Отклонено `0`** — телефон до компьютера не дошёл. Дело не в TGLock: проверь, что оба устройства в одной сети (телефон может сидеть на гостевом Wi-Fi или в мобильном интернете), что в роутере не включена изоляция клиентов, и что брандмауэр пускает входящие на порт TGLock.
> - **Соединения растут, Отклонено растёт** — телефон дошёл, но просит адрес, который LAN-режим не пропускает. Конкретный адрес назван в журнале событий ниже — пришли эту строку в issue.
> - **Не опознаны растёт** — телефон дошёл, но договориться не вышло. Почти всегда в Telegram на телефоне вписана ссылка от прошлого запуска, то есть другой секрет. Сверь её с той, что показана в окне сейчас.
> - **Соединения растут, Туннели `0`** — до Telegram не доходит уже сам компьютер. Это [Cloudflare Worker](docs/CLOUDFLARE_WORKER.md), а не проблема LAN.
### 🖥 Без графического интерфейса: `tglock-cli`
@@ -207,7 +208,7 @@ worker = ["my-name.workers.dev"]
Файл с секретом внутри держите с правами `600`: это доступ к вашему прокси.
При запуске печатается готовая `tg://proxy`-ссылка — её можно открыть на любом устройстве в сети, чтобы Telegram настроился сам. Дальше в лог идёт по строке на каждое изменение состояния: сколько соединений, какой дата-центр, какой маршрут живой, сколько сбоев и сколько запросов отклонено политикой «только Telegram». Отдельными строками отмечаются подключившиеся устройства и адреса, из-за которых был отказ, — по ним видно, дошёл ли телефон до сервиса вообще.
При запуске печатается готовая `tg://proxy`-ссылка — её можно открыть на любом устройстве в сети, чтобы Telegram настроился сам. Дальше в лог идёт по строке на каждое изменение состояния: сколько соединений, какой дата-центр, какой маршрут живой, сколько сбоев, сколько запросов отклонено политикой «только Telegram» и сколько клиентов не опознано. Отдельными строками отмечаются подключившиеся устройства, адреса, из-за которых был отказ, и клиенты, с которыми не удалось договориться, — по ним видно, дошёл ли телефон до сервиса вообще и не вписан ли в нём устаревший секрет.
Прав администратора не нужно: TGLock не правит ни системный DNS, ни файл `hosts` — нужные адреса Telegram зашиты в маршрутах, а TLS SNI остаётся настоящим.
+97
View File
@@ -97,6 +97,83 @@ LAN-режим превращал бы машину в открытый прок
и дошёл, но попросил адрес, который мы не пропускаем. Первый виден как ноль
соединений и ноль отказов, второй — как соединения есть, отказы растут.
Третий случай нашёлся, когда репортёр #42 прислал диагностику: у него было ноль
отказов и работающие туннели, то есть оба счётчика говорили «всё хорошо».
Клиент, который дошёл до прокси, но не сумел договориться, не попадал ни в
один из них. Соединение просто закрывалось: `active` дёргался вверх и обратно.
Теперь такие клиенты считает `unknown_clients`, и журнал называет адрес и
причину. Их две:
- MTProto-init не разбирается под текущим секретом. Почти всегда это ссылка
`tg://proxy` от прошлого запуска: секрет — её половина, и клиент со
сохранённой старой ссылкой попадает ровно сюда. Со стороны Telegram это и
есть «прокси настроен неверно и будет отключён» (#37).
- SOCKS5-приветствие не разбирается. Сюда же попадает MTProto-соединение,
ушедшее в SOCKS5-ветку по неоднозначному первому байту, если полный init не
успел прийти за `PROTOCOL_PROBE_TIMEOUT`.
Счётчик `ws_failures` от них отличается тем, что растёт после успешного
рукопожатия с клиентом: там договорились с клиентом, но не смогли с Telegram.
### Почему не поднялся туннель
`ws_failures` говорит, что каскад маршрутов упал целиком, и молчит о причине.
Текст с перечислением попыток собирался в `TransportEngine::connect` и там же
пропадал: наверх уходил `Err`, который выбрасывался в `serve`. При `туннелей 0`
и растущих сбоях отличить «провайдер режет закреплённые адреса» от «воркер
отвечает отказом» было нечем — ровно та стена, в которую упёрся репортёр #50.
Теперь причина попадает в журнал одной строкой на каждый набор отказов:
```
Не поднялся туннель до DC2: 149.154.167.51 — не отвечает (таймаут TCP);
kws2.web.telegram.org — таймаут TLS/WebSocket; my.workers.dev — рукопожатие
WebSocket: HTTP error: 403 Forbidden
```
Дедупликация журнала делает эту строку разовой: маршруты у DC стабильны, и
повтор той же комбинации отказов не пишется.
Домены Cloudflare Worker отчитываются так же. Строка, не похожая на имя хоста,
раньше отбрасывалась молча — `https://name.workers.dev/` со схемой или слэшем не
проходит `valid_domain`, маршрут не появлялся, и «воркер настроен» ничем не
отличалось от «воркера нет». Теперь отвергнутая строка называется вместе с
причиной, а принятая подтверждается: `Cloudflare Worker в списке маршрутов:
name.workers.dev`.
## Туннель: два независимых направления
Каждое клиентское соединение получает свой WebSocket-туннель, и внутри него
данные идут в обе стороны сразу. До 2.0.0-beta.12 оба направления обслуживал
один `select!` с пометкой `biased`, и это давало два дефекта, снаружи
выглядевших одинаково: «Подключено», а ничего не идёт.
`biased` опрашивает ветки строго по порядку. Пока в первой — «Telegram →
клиент» — есть данные, до второй очередь не доходит вообще. То есть при
непрерывном потоке вниз (первичная синхронизация телефона, загрузка медиа)
исходящие пакеты клиента не читались.
Второй дефект — одна задача на оба направления. `tcp_w.write_all` ждёт, пока
клиент разберёт присланное, и всё это время не опрашивается чтение от клиента.
Телефон по Wi-Fi разбирает поток медленнее, чем Telegram Desktop на той же
машине через loopback, — отсюда асимметрия «на компьютере работает, на телефоне
нет» из #42.
Для MTProto это фатально: клиент обязан слать подтверждения, а за каждым
следующим куском файла — свой `upload.getFile`. Первый запрос уходит, дальше
идёт поток вниз, и следующие запросы наверх не попадают. Загрузка встаёт при
живом туннеле, нулевых сбоях и нулевых отклонениях — ровно картина из #32.
Теперь это две независимые половины: `ws.split()` плюс `CryptoContext::split()`,
потому что шифры направлений независимы — два потока AES-CTR со своими ключами.
Ping приходит в читающую половину, а отвечает на него пишущая, через канал на
четыре слота: владелец отправляющей половины должен оставаться ровно один.
Оба дефекта закрыты тестами, которые падают на beta.11. Первый: за пять секунд
непрерывной загрузки наверх не уходит ни одного байта. Второй: клиент, не
успевающий читать, замораживает собственную отправку.
## Учёт состояния
`Stats::ws` считает **установленные** туннели: счётчик поднимается после
@@ -105,6 +182,14 @@ LAN-режим превращал бы машину в открытый прок
несколько секунд каждый. Состояния «порт открыт», «идёт перебор маршрутов» и
«туннель установлен» различимы и в GUI, и в выводе CLI.
Дата-центр и маршрут пишутся **одним значением**, в момент, когда туннель
поднялся. Пока это были два независимых поля, номер писало соединение при
разборе init, а маршрут — другое соединение после рукопожатия, и при десятках
одновременных соединений в строку статуса попадала пара из разных из них.
Читалась она как «до этого DC шли этим маршрутом», хотя означала другое: в
диагностике #42 встречались строки `DC5 · Запасной Telegram IP`, а у DC5
закреплённый адрес всего один и запасного у него не бывает вовсе.
Секрет прокси — половина ссылки `tg://proxy`. Для сервиса его нужно закрепить
файлом (`--secret-file`): под `DynamicUser` и `ProtectHome` домашней папки нет,
путь по умолчанию не определяется, и секрет генерировался бы заново при каждом
@@ -148,6 +233,14 @@ Worker должен принимать WebSocket на:
добавить собственную авторизацию до стабильного релиза; поэтому Worker
остаётся расширенной опцией alpha-версии.
Запись в сокет Telegram обязана быть последовательной: следующий чанк уходит
после того, как записан предыдущий, и только когда писатель к этому готов
(`writer.ready`). В `worker/tglock-worker.js` этого не было — `write()`
вызывался поверх незавершённого, без backpressure. Пока в клиенте голодала
отправка, настоящего потока вверх через воркер не возникало и это не
проявлялось; после того как голодание починили, поток появился. Кто разворачивал
воркер раньше — обновите скрипт.
## Current limitations
- SNI camouflage не включена: небезопасное отключение hostname verification
@@ -170,5 +263,9 @@ Worker должен принимать WebSocket на:
CLI собирается одной командой. Подробно — в разделе README про антивирус.
- Работоспособность медиа зависит от конкретного DC аккаунта и доступности
Telegram/Cloudflare у провайдера.
- Соединение, открытое клиентом и молчащее дольше `IO_TIMEOUT` (10 секунд),
закрывается. Для клиента, открывающего соединения про запас, это норма; счётчик
«промолчали» показывает, как часто это происходит, — раньше такие соединения
не попадали никуда.
- Пулы заранее открытых WebSocket-соединений будут добавлены после измерения,
что они не создают лишнюю нагрузку и не ухудшают стабильность.
+2
View File
@@ -41,6 +41,8 @@
Если вернулось `not found` — проверь, что путь именно `/apiws`. Если ошибка про `cloudflare:sockets` — у воркера слишком старая дата совместимости, поставь в **Settings → Compatibility date** сегодняшнюю.
> **Разворачивал воркер до 2.0.0-beta.14 — обнови скрипт.** В прежней версии запись в сокет Telegram шла без ожидания предыдущей и без backpressure. Пока в клиенте голодала отправка, через воркер не проходило настоящего потока вверх и это не проявлялось; после того как голодание починили в beta.12, поток появился.
## Подключение в TGLock
**В приложении:** Настройки → поле **Cloudflare Worker** → вставь `tglock.имя.workers.dev` → Сохранить. Настройки меняются только при выключенной защите.
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "tglock-ui",
"private": true,
"version": "2.0.0-beta.9",
"version": "2.0.0-beta.14",
"type": "module",
"scripts": {
"dev": "vite --port 1420",
+7 -4
View File
@@ -218,19 +218,22 @@ async fn watch_status(stats: Arc<proxy::Stats>) {
let current = (
stats.active.load(Ordering::Relaxed),
stats.ws.load(Ordering::Relaxed),
stats.last_dc.load(Ordering::Relaxed),
stats.last_route.load(Ordering::Relaxed),
stats.last_dc(),
stats.last_route(),
stats.ws_failures.load(Ordering::Relaxed),
stats.route_failures(),
stats.blocked.load(Ordering::Relaxed),
stats.unknown_clients.load(Ordering::Relaxed),
stats.silent_clients.load(Ordering::Relaxed),
);
if previous.as_ref() == Some(&current) {
continue;
}
let (active, tunnels, dc, route, failures, route_failures, blocked) = current;
let (active, tunnels, dc, route, failures, route_failures, blocked, unknown, silent) =
current;
let line = format!(
"соединений {active} · туннелей {tunnels} · {} · {} · сбоев {failures} · \
падений маршрутов {route_failures} · отклонено {blocked}",
падений маршрутов {route_failures} · отклонено {blocked} · не опознано {unknown} · промолчали {silent}",
if dc > 0 {
format!("DC{dc}")
} else {
+13 -2
View File
@@ -53,6 +53,15 @@ struct StatusSnapshot {
/// этой машины; не ноль — что дотянулся, и разбираться надо с адресами
/// (by-sonic/tglock#42).
blocked: u32,
/// Клиенты, которые дошли, но не сумели договориться. Почти всегда это
/// ссылка `tg://proxy` от прошлого запуска, то есть другой секрет.
unknown_clients: u32,
/// Соединения, которые открылись и ничего не прислали до таймаута.
///
/// Растущее число в такт с переподключениями клиента означает, что он
/// открывает соединения впрок, а мы закрываем их по таймауту
/// (by-sonic/tglock#42).
silent_clients: u32,
uptime_seconds: u64,
port: u16,
/// Адрес, который нужно вписать в Telegram на другом устройстве.
@@ -113,8 +122,8 @@ impl AppState {
for event in self.stats.drain_events() {
self.log(event, false);
}
let data_center = self.stats.last_dc.load(Ordering::Relaxed);
let route = transport::route_label(self.stats.last_route.load(Ordering::Relaxed));
let data_center = self.stats.last_dc();
let route = transport::route_label(self.stats.last_route());
StatusSnapshot {
running: self.stats.running.load(Ordering::SeqCst),
active_connections: self.stats.active.load(Ordering::Relaxed),
@@ -124,6 +133,8 @@ impl AppState {
failures: self.stats.ws_failures.load(Ordering::Relaxed),
route_failures: self.stats.route_failures(),
blocked: self.stats.blocked.load(Ordering::Relaxed),
unknown_clients: self.stats.unknown_clients.load(Ordering::Relaxed),
silent_clients: self.stats.silent_clients.load(Ordering::Relaxed),
uptime_seconds: self
.started_at
.lock()
+43
View File
@@ -43,6 +43,49 @@ impl CryptoContext {
self.telegram_decrypt.apply_keystream(data);
self.client_encrypt.apply_keystream(data);
}
/// Разделить шифры по направлениям, чтобы туннель шёл в обе стороны сразу.
///
/// Направления независимы: это два потока AES-CTR со своими ключами, и ни
/// один байт одного не влияет на другой.
pub fn split(self) -> (Upstream, Downstream) {
(
Upstream {
client_decrypt: self.client_decrypt,
telegram_encrypt: self.telegram_encrypt,
},
Downstream {
telegram_decrypt: self.telegram_decrypt,
client_encrypt: self.client_encrypt,
},
)
}
}
/// Шифры направления «клиент -> Telegram».
pub struct Upstream {
client_decrypt: AesCtr,
telegram_encrypt: AesCtr,
}
impl Upstream {
pub fn apply(&mut self, data: &mut [u8]) {
self.client_decrypt.apply_keystream(data);
self.telegram_encrypt.apply_keystream(data);
}
}
/// Шифры направления «Telegram -> клиент».
pub struct Downstream {
telegram_decrypt: AesCtr,
client_encrypt: AesCtr,
}
impl Downstream {
pub fn apply(&mut self, data: &mut [u8]) {
self.telegram_decrypt.apply_keystream(data);
self.client_encrypt.apply_keystream(data);
}
}
pub fn generate_secret() -> [u8; 16] {
+699 -62
View File
@@ -1,7 +1,7 @@
use crate::config::ListenConfig;
use std::collections::{HashSet, VecDeque};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, AtomicU8, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
@@ -16,13 +16,17 @@ const SOCKS5_VERSION: u8 = 0x05;
/// byte as the start of a SOCKS5 greeting.
const PROTOCOL_PROBE_TIMEOUT: Duration = Duration::from_millis(250);
const PROTOCOL_PROBE_INTERVAL: Duration = Duration::from_millis(5);
/// Сколько неотвеченных Ping'ов держать, пока отправляющая половина занята.
const PONG_QUEUE: usize = 4;
pub struct Stats {
pub running: AtomicBool,
pub active: AtomicU32,
pub total: AtomicU32,
pub ws: AtomicU32,
pub last_dc: AtomicU16,
/// DC последнего разобранного соединения. Показывается, пока туннеля ещё
/// нет: клиент уже понят, маршрут ещё не выбран.
seen_dc: AtomicU16,
pub ws_failures: AtomicU32,
/// Сколько запросов отклонено политикой «только Telegram».
///
@@ -31,8 +35,34 @@ pub struct Stats {
/// Раньше отказ происходил молча, и снаружи оба случая выглядели одинаково
/// (by-sonic/tglock#42).
pub blocked: AtomicU32,
/// See `transport::RouteKind::ui_code`.
pub last_route: AtomicU8,
/// Сколько клиентов дошло до прокси, но не сумело договориться.
///
/// Почти всегда это несовпадение секрета: в Telegram вписана ссылка от
/// прошлого запуска. Такое соединение закрывалось молча, и по диагностике
/// отличить его от рабочего было нельзя.
pub unknown_clients: AtomicU32,
/// Сколько соединений открылось и ничего не прислало за `IO_TIMEOUT`.
///
/// Такое соединение закрывается по таймауту и до сих пор не попадало ни в
/// один счётчик: `unknown_clients` растёт, только когда запрос пришёл и не
/// разобрался, а не когда его не дождались. По диагностике это выглядело
/// как соединение без туннеля и ничего больше — а репортёр #42 видел, что
/// его телефон переустанавливает соединение примерно раз в десять секунд,
/// то есть ровно с этим периодом.
pub silent_clients: AtomicU32,
/// DC и маршрут последнего поднятого туннеля, упакованные в одно значение.
///
/// Раньше это были два независимых поля: номер писало соединение при
/// разборе init, маршрут — другое соединение после рукопожатия. При
/// нескольких десятках одновременных соединений пара в строке статуса
/// складывалась из разных из них, и читалась она как «до этого DC шли
/// этим маршрутом», хотя означала совсем не это. У DC1, DC3, DC5 и DC203
/// закреплённый адрес всего один, и «запасного» у них не бывает вовсе —
/// а строки `DC5 · Запасной Telegram IP` в диагностике встречались
/// (by-sonic/tglock#42).
///
/// Формат: `dc << 8 | route`, где route — `transport::RouteKind::ui_code`.
last_tunnel: AtomicU32,
transport: crate::transport::TransportEngine,
secret: [u8; 16],
/// Почему секрет не удалось сохранить, если не удалось.
@@ -88,10 +118,12 @@ impl Stats {
active: AtomicU32::new(0),
total: AtomicU32::new(0),
ws: AtomicU32::new(0),
last_dc: AtomicU16::new(0),
seen_dc: AtomicU16::new(0),
ws_failures: AtomicU32::new(0),
blocked: AtomicU32::new(0),
last_route: AtomicU8::new(0),
unknown_clients: AtomicU32::new(0),
silent_clients: AtomicU32::new(0),
last_tunnel: AtomicU32::new(0),
transport: crate::transport::TransportEngine::new(),
secret,
secret_write_error,
@@ -122,6 +154,39 @@ impl Stats {
));
}
/// Клиент дошёл, но договориться с ним не удалось.
///
/// Раньше такое соединение закрывалось молча: `active` дёргался вверх и
/// обратно, и всё. По диагностике это неотличимо от «клиент подключился и
/// работает», хотя означает противоположное (by-sonic/tglock#42).
fn note_unknown_client(&self, peer: Option<SocketAddr>, reason: &str) {
self.unknown_clients.fetch_add(1, Ordering::Relaxed);
let who = match peer {
Some(peer) => peer.ip().to_string(),
None => "неизвестный адрес".to_owned(),
};
self.note(format!("Клиент {who}: {reason}"));
}
/// Клиент открыл соединение и не сказал ничего.
///
/// Отличается от `note_unknown_client` тем, что там запрос пришёл и не
/// разобрался, а здесь его не дождались. Для клиента, который открывает
/// соединения про запас, это норма; для клиента, который переоткрывает их
/// в такт с таймаутом, — нет, и разницу видно только по счётчику
/// (by-sonic/tglock#42).
fn note_silent_client(&self, peer: Option<SocketAddr>, reason: &str) {
self.silent_clients.fetch_add(1, Ordering::Relaxed);
let who = match peer {
Some(peer) => peer.ip().to_string(),
None => "неизвестный адрес".to_owned(),
};
self.note(format!(
"Клиент {who}: {reason} за {} с — соединение закрыто",
IO_TIMEOUT.as_secs()
));
}
/// Отметить, что до прокси дотянулось устройство из сети, а не с этой машины.
///
/// Это первое, что нужно знать при разборе LAN-режима: если строки нет,
@@ -133,6 +198,34 @@ impl Stats {
self.note(format!("Подключилось устройство из сети: {}", peer.ip()));
}
/// Запомнить DC, с которым пришёл клиент. Туннеля может ещё не быть.
fn note_dc(&self, dc: u16) {
self.seen_dc.store(dc, Ordering::Relaxed);
}
/// Запомнить, каким маршрутом поднялся туннель и до какого DC.
///
/// Пишется одним значением, чтобы пара в диагностике всегда была из
/// одного соединения.
fn note_tunnel(&self, dc: u16, route: u8) {
self.last_tunnel
.store(u32::from(dc) << 8 | u32::from(route), Ordering::Relaxed);
}
/// Номер дата-центра для показа: из последнего поднятого туннеля, а пока
/// туннеля не было — из последнего разобранного соединения.
pub fn last_dc(&self) -> u16 {
match self.last_tunnel.load(Ordering::Relaxed) {
0 => self.seen_dc.load(Ordering::Relaxed),
packed => (packed >> 8) as u16,
}
}
/// Маршрут последнего поднятого туннеля. См. `transport::RouteKind::ui_code`.
pub fn last_route(&self) -> u8 {
(self.last_tunnel.load(Ordering::Relaxed) & 0xff) as u8
}
pub fn telegram_secret(&self) -> String {
crate::mtproto::telegram_secret(&self.secret)
}
@@ -153,7 +246,18 @@ impl Stats {
.filter(|value| !value.trim().is_empty())
.map(str::to_owned)
.collect::<Vec<_>>();
self.transport.set_worker_domains(&domains);
let result = self.transport.set_worker_domains(&domains);
// Молчание здесь неотличимо от «воркер работает»: пока строка не
// попадала в список маршрутов, об этом не сообщалось ничем, и человек
// считал резервный маршрут настроенным (by-sonic/tglock#50).
for rejected in &result.rejected {
self.note(format!(
"Cloudflare Worker «{rejected}» не похож на имя хоста — маршрут не добавлен. Нужно только имя, без https:// и без косой черты: example.workers.dev"
));
}
for accepted in &result.accepted {
self.note(format!("Cloudflare Worker в списке маршрутов: {accepted}"));
}
}
pub fn stop(&self) {
@@ -274,9 +378,13 @@ async fn detect_protocol(
stats: &Stats,
) -> Result<Protocol, Box<dyn std::error::Error + Send + Sync>> {
let mut probe = [0; INIT_LEN];
let peeked = tokio::time::timeout(IO_TIMEOUT, stream.peek(&mut probe[..1]))
.await
.map_err(|_| "client protocol detection timeout")??;
let peeked = match tokio::time::timeout(IO_TIMEOUT, stream.peek(&mut probe[..1])).await {
Ok(peeked) => peeked?,
Err(_) => {
stats.note_silent_client(stream.peer_addr().ok(), "не прислал ни байта");
return Err("client protocol detection timeout".into());
}
};
if peeked == 0 {
return Ok(Protocol::Empty);
}
@@ -335,10 +443,19 @@ async fn handle_socks5(
allow_direct: bool,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
s.set_nodelay(true)?;
let peer = s.peer_addr().ok();
let (addr, port) = tokio::time::timeout(IO_TIMEOUT, read_socks5_request(&mut s))
let request = tokio::time::timeout(IO_TIMEOUT, read_socks5_request(&mut s))
.await
.map_err(|_| "SOCKS5 handshake timeout")??;
.map_err(|_| "SOCKS5 handshake timeout".to_owned())
.and_then(|result| result.map_err(|error| error.to_string()));
let (addr, port) = match request {
Ok(request) => request,
Err(error) => {
stats.note_unknown_client(peer, &format!("SOCKS5-приветствие не разобрано ({error})"));
return Err(error.into());
}
};
let destination = classify(&addr);
if destination == Destination::Elsewhere && !allow_direct {
stats.note_blocked(&addr, port);
@@ -358,7 +475,7 @@ async fn handle_socks5(
let (dc, media) = dc_from_init(&init)
.unwrap_or_else(|| (crate::telegram_net::dc_from_ip(ip).unwrap_or(2), false));
stats.last_dc.store(dc, Ordering::Relaxed);
stats.note_dc(dc);
let r = ws_tunnel(s, dc, media, &init, None, stats).await;
@@ -436,14 +553,33 @@ async fn handle_mtproto(
stats: &Stats,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
stream.set_nodelay(true)?;
let peer = stream.peer_addr().ok();
let mut init = [0; 64];
tokio::time::timeout(IO_TIMEOUT, stream.read_exact(&mut init))
.await
.map_err(|_| "MTProto init timeout")??;
let parsed = crate::mtproto::parse_client_init(&init, &stats.secret)
.ok_or("invalid MTProto init or secret")?;
match tokio::time::timeout(IO_TIMEOUT, stream.read_exact(&mut init)).await {
Ok(result) => {
result?;
}
Err(_) => {
stats.note_silent_client(peer, "начал MTProto-init и не дослал его");
return Err("MTProto init timeout".into());
}
}
let parsed = match crate::mtproto::parse_client_init(&init, &stats.secret) {
Some(parsed) => parsed,
None => {
// Секрет — половина ссылки `tg://proxy`. Клиент с сохранённой
// ссылкой от прошлого запуска попадает ровно сюда, и Telegram
// показывает ему «прокси настроен неверно».
stats.note_unknown_client(
peer,
"MTProto-init не разобран. Скорее всего в Telegram вписан другой \
секрет сверьте ссылку tg://proxy с той, что показана сейчас",
);
return Err("invalid MTProto init or secret".into());
}
};
stats.last_dc.store(parsed.dc, Ordering::Relaxed);
stats.note_dc(parsed.dc);
let result = ws_tunnel(
stream,
parsed.dc,
@@ -574,57 +710,102 @@ async fn ws_tunnel(
dc: u16,
media: bool,
init: &[u8; 64],
mut crypto: Option<crate::mtproto::CryptoContext>,
crypto: Option<crate::mtproto::CryptoContext>,
stats: &Stats,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
use futures_util::{SinkExt, StreamExt};
let (mut ws, connected) = stats.transport.connect(dc, media).await?;
let (ws, connected) = match stats.transport.connect(dc, media).await {
Ok(connected) => connected,
Err(error) => {
// Единственное место, где известно, ПОЧЕМУ туннеля нет. Раньше
// текст уходил в `Err` и там пропадал: оставался счётчик сбоев без
// причины, и отличить «провайдер режет адреса» от «воркер отвечает
// отказом» было нечем (by-sonic/tglock#50).
stats.note(error.clone());
return Err(error.into());
}
};
let _tunnel = EstablishedTunnel::new(stats);
stats
.last_route
.store(connected.route.kind.ui_code(), Ordering::Relaxed);
stats.note_tunnel(dc, connected.route.kind.ui_code());
let (mut tcp_r, mut tcp_w) = tokio::io::split(tcp);
let (mut ws_w, mut ws_r) = ws.split();
let (upstream_crypto, downstream_crypto) = match crypto.map(|crypto| crypto.split()) {
Some((upstream, downstream)) => (Some(upstream), Some(downstream)),
None => (None, None),
};
// Send buffered init as first frame
ws.send(tungstenite::Message::Binary(init.to_vec())).await?;
ws_w.send(tungstenite::Message::Binary(init.to_vec()))
.await?;
let mut buf = vec![0u8; 65536];
// Ping приходит в половину, которая читает, а отвечать на него должна та,
// которая пишет: владелец у отправляющей половины строго один.
let (pong_tx, mut pong_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(PONG_QUEUE);
loop {
tokio::select! {
biased;
msg = ws.next() => match msg {
Some(Ok(tungstenite::Message::Binary(mut data))) => {
// Направления работают независимо друг от друга. Раньше это был один
// `select!`, и любое ожидание внутри него останавливало вторую половину:
// непрерывная загрузка не давала опросить клиента вообще, а клиент,
// который не успевал разбирать входящий поток, замораживал заодно и свою
// отправку. Telegram при этом ждёт от клиента подтверждений — без них
// сессия встаёт при живом туннеле (by-sonic/tglock#42, #32).
let downstream = async {
let mut crypto = downstream_crypto;
while let Some(message) = ws_r.next().await {
match message {
Ok(tungstenite::Message::Binary(mut data)) => {
if let Some(crypto) = &mut crypto {
crypto.telegram_to_client(data.as_mut());
crypto.apply(data.as_mut());
}
tcp_w.write_all(data.as_ref()).await?;
tcp_w.flush().await?;
}
Some(Ok(tungstenite::Message::Ping(p))) => {
let _ = ws.send(tungstenite::Message::Pong(p)).await;
}
Some(Ok(tungstenite::Message::Close(_))) | None => break,
Some(Err(_)) => break,
_ => {}
},
n = tcp_r.read(&mut buf) => match n {
Ok(0) | Err(_) => break,
Ok(n) => {
if let Some(crypto) = &mut crypto {
crypto.client_to_telegram(&mut buf[..n]);
Ok(tungstenite::Message::Ping(payload)) => {
if pong_tx.send(payload).await.is_err() {
break;
}
ws.send(tungstenite::Message::Binary(buf[..n].to_vec())).await?;
}
},
Ok(tungstenite::Message::Close(_)) | Err(_) => break,
Ok(_) => {}
}
}
}
Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
};
let _ = ws.close(None).await;
let upstream = async {
let mut crypto = upstream_crypto;
let mut buf = vec![0u8; 65536];
loop {
tokio::select! {
read = tcp_r.read(&mut buf) => match read {
Ok(0) | Err(_) => break,
Ok(read) => {
if let Some(crypto) = &mut crypto {
crypto.apply(&mut buf[..read]);
}
ws_w
.send(tungstenite::Message::Binary(buf[..read].to_vec()))
.await?;
}
},
payload = pong_rx.recv() => match payload {
Some(payload) => {
ws_w.send(tungstenite::Message::Pong(payload)).await?;
}
None => break,
},
}
}
let _ = ws_w.close().await;
Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
};
tokio::pin!(downstream, upstream);
tokio::select! {
result = &mut downstream => result?,
result = &mut upstream => result?,
}
Ok(())
}
@@ -859,6 +1040,85 @@ mod tests {
);
}
/// Диагностика обязана показывать пару из одного соединения.
///
/// Пока это были два независимых поля, при десятках одновременных
/// соединений в строку статуса попадали номер от одного и маршрут от
/// другого. Читалось это как «до DC5 шли запасным адресом», хотя у DC5
/// закреплённый адрес всего один и запасного не бывает вовсе
/// (by-sonic/tglock#42).
#[test]
fn the_reported_data_centre_and_route_come_from_the_same_tunnel() {
let stats = Stats::new();
assert_eq!(stats.last_dc(), 0, "до соединений показывать нечего");
assert_eq!(stats.last_route(), 0);
stats.note_dc(2);
assert_eq!(stats.last_dc(), 2, "клиент разобран, номер известен");
assert_eq!(stats.last_route(), 0, "а маршрут ещё не выбран");
stats.note_tunnel(4, 2);
assert_eq!((stats.last_dc(), stats.last_route()), (4, 2));
// Ещё одно соединение до другого DC, туннеля у него пока нет.
stats.note_dc(203);
assert_eq!(
(stats.last_dc(), stats.last_route()),
(4, 2),
"пара обязана остаться от соединения, у которого туннель был"
);
}
/// Клиент с сохранённой ссылкой от прошлого запуска. Раньше его соединение
/// закрывалось молча, и по диагностике это было неотличимо от рабочего.
#[tokio::test]
async fn a_client_with_the_wrong_secret_gets_named_instead_of_dropped_in_silence() {
let stats = Stats::new();
let (port, server) = start_proxy(stats.clone(), true).await;
let stranger = crate::mtproto::generate_secret();
let init = unambiguous_client_init(&stranger, 2);
let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
client.write_all(&init).await.unwrap();
wait_until("отказ по секрету", || {
stats.unknown_clients.load(Ordering::Relaxed) > 0
})
.await;
let events = stats.drain_events();
assert!(
events
.iter()
.any(|event| event.contains("MTProto-init не разобран")),
"в журнале должно быть сказано, что init не разобран: {events:?}"
);
stats.stop();
let _ = server.await.unwrap();
}
#[tokio::test]
async fn a_client_that_speaks_neither_protocol_is_counted_too() {
let stats = Stats::new();
let (port, server) = start_proxy(stats.clone(), true).await;
// Приветствие SOCKS5 с нулём методов: разбор обязан провалиться.
let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
client.write_all(&[0x05, 0x00]).await.unwrap();
wait_until("отказ по рукопожатию", || {
stats.unknown_clients.load(Ordering::Relaxed) > 0
})
.await;
assert!(stats
.drain_events()
.iter()
.any(|event| event.contains("SOCKS5-приветствие не разобрано")));
stats.stop();
let _ = server.await.unwrap();
}
#[tokio::test]
async fn stops_listener_and_active_tasks_cleanly() {
let reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
@@ -987,7 +1247,7 @@ mod tests {
// Detection must land on MTProto, which records the data centre. The
// SOCKS5 path would instead answer with a handshake reply.
wait_until("the MTProto data centre to be recorded", || {
stats.last_dc.load(Ordering::Relaxed) == 2
stats.last_dc() == 2
})
.await;
@@ -1140,9 +1400,9 @@ mod tests {
"Telegram must receive exactly the client's plaintext"
);
assert_eq!(stats.last_dc.load(Ordering::Relaxed), 4);
assert_eq!(stats.last_dc(), 4);
assert_eq!(
stats.last_route.load(Ordering::Relaxed),
stats.last_route(),
crate::transport::RouteKind::TelegramIp.ui_code()
);
assert_eq!(stats.ws_failures.load(Ordering::Relaxed), 0);
@@ -1197,7 +1457,7 @@ mod tests {
);
assert_eq!(relayed, request);
assert_eq!(
stats.last_route.load(Ordering::Relaxed),
stats.last_route(),
crate::transport::RouteKind::CloudflareWorker.ui_code()
);
@@ -1225,10 +1485,7 @@ mod tests {
let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
client.write_all(&init).await.unwrap();
wait_until("the init to be parsed", || {
stats.last_dc.load(Ordering::Relaxed) == 2
})
.await;
wait_until("the init to be parsed", || stats.last_dc() == 2).await;
tokio::time::sleep(Duration::from_millis(300)).await;
assert_eq!(
@@ -1237,7 +1494,7 @@ mod tests {
"a handshake still in flight must not be reported as a working tunnel"
);
assert_eq!(
stats.last_route.load(Ordering::Relaxed),
stats.last_route(),
0,
"no route may be announced before a tunnel is established"
);
@@ -1266,7 +1523,7 @@ mod tests {
})
.await;
assert_eq!(
stats.last_route.load(Ordering::Relaxed),
stats.last_route(),
0,
"a route must not be reported as working when every attempt failed"
);
@@ -1276,6 +1533,208 @@ mod tests {
let _ = server.await.unwrap();
}
/// Диагностика обязана называть причину, а не только считать сбои.
///
/// При `туннелей 0` счётчик сбоев говорит, что не получилось, и молчит о
/// том, почему. Текст ошибки собирался и выбрасывался, и разобрать
/// «провайдер режет адреса» против «воркер отвечает отказом» было нечем
/// (by-sonic/tglock#50).
#[tokio::test]
async fn a_cascade_that_failed_says_why_in_the_log() {
let dead = TcpListener::bind("127.0.0.1:0").await.unwrap();
let dead_port = dead.local_addr().unwrap().port();
drop(dead);
let stats = Stats::new();
stats.transport.force_local_route(dead_port);
let (port, server) = start_proxy(stats.clone(), false).await;
let init = unambiguous_client_init(&stats.secret, 2);
let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
client.write_all(&init).await.unwrap();
wait_until("сбой засчитан", || {
stats.ws_failures.load(Ordering::Relaxed) > 0
})
.await;
let events = stats.drain_events();
let named = events
.iter()
.find(|event| event.contains("Не поднялся туннель до DC2"))
.unwrap_or_else(|| panic!("причина отказа не попала в журнал: {events:?}"));
assert!(
named.contains("127.0.0.1"),
"в журнале должен быть назван адрес, до которого не дошли: {named}"
);
stats.stop();
let _ = server.await.unwrap();
}
/// Строка, не похожая на имя хоста, отбрасывалась молча, и «воркер
/// настроен» ничем не отличалось от «воркера нет» (by-sonic/tglock#50).
#[test]
fn a_worker_domain_is_confirmed_or_named_as_rejected() {
let stats = Stats::new();
stats.set_worker_domain("https://mine.workers.dev/, spare.workers.dev");
let events = stats.drain_events();
assert!(
events
.iter()
.any(|event| event.contains("https://mine.workers.dev/")
&& event.contains("не похож на имя хоста")),
"отвергнутый домен должен быть назван вместе с причиной: {events:?}"
);
assert!(
events
.iter()
.any(|event| event.contains("в списке маршрутов: spare.workers.dev")),
"принятый домен нужно подтвердить, иначе проверить нечем: {events:?}"
);
}
/// Ping от той стороны обязан получить Pong, и туннель обязан это пережить.
///
/// Ping приходит в читающую половину, а отвечать на него должна пишущая.
/// Единственный маршрут, где Ping вообще бывает, — Cloudflare Worker:
/// у Telegram его нет. Поэтому поломка на этом пути видна только тем, у
/// кого настроен воркер (by-sonic/tglock#42).
#[allow(clippy::result_large_err)]
#[tokio::test]
async fn a_ping_is_answered_and_the_tunnel_survives_it() {
use futures_util::{SinkExt, StreamExt};
let relay_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let relay_port = relay_listener.local_addr().unwrap().port();
let (report_tx, report_rx) = tokio::sync::oneshot::channel::<Result<Vec<u8>, String>>();
tokio::spawn(async move {
let (tcp, _) = relay_listener.accept().await.unwrap();
let mut websocket =
tokio_tungstenite::accept_hdr_async(tcp, |_: &Request, mut response: Response| {
response.headers_mut().insert(
"Sec-WebSocket-Protocol",
"binary".parse().expect("static header value"),
);
Ok(response)
})
.await
.unwrap();
let init = match websocket.next().await {
Some(Ok(Message::Binary(data))) => data,
other => {
let _ = report_tx.send(Err(format!("ожидался init, пришло {other:?}")));
return;
}
};
let header: [u8; INIT_LEN] = init.as_slice().try_into().unwrap();
let mut relay = crate::mtproto::test_relay_peer(&header);
websocket
.send(Message::Ping(b"keepalive".to_vec()))
.await
.unwrap();
let mut answered = false;
let mut payload = Vec::new();
while payload.is_empty() {
match websocket.next().await {
Some(Ok(Message::Pong(pong))) => {
answered = pong == b"keepalive";
}
Some(Ok(Message::Binary(mut data))) => {
relay.decrypt(&mut data);
payload.extend_from_slice(&data);
}
Some(Ok(_)) => {}
other => {
let _ = report_tx.send(Err(format!(
"туннель закрылся до полезных данных: {other:?}"
)));
return;
}
}
}
let _ = report_tx.send(if answered {
Ok(payload)
} else {
Err("Pong не пришёл".to_owned())
});
});
let stats = Stats::new();
stats.transport.force_local_route(relay_port);
let (port, server) = start_proxy(stats.clone(), false).await;
let init = unambiguous_client_init(&stats.secret, -4);
let mut peer = crate::mtproto::test_client_peer(&init, &stats.secret);
let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
client.write_all(&init).await.unwrap();
// Клиент шлёт после Ping'а: если ответ на Ping ломает пишущую половину,
// это не дойдёт.
tokio::time::sleep(Duration::from_millis(200)).await;
let request = b"a request sent after the ping".to_vec();
let mut wire = request.clone();
peer.encrypt(&mut wire);
client.write_all(&wire).await.unwrap();
let report = tokio::time::timeout(Duration::from_secs(5), report_rx)
.await
.expect("реле должно доложить о результате")
.unwrap();
assert_eq!(report, Ok(request), "Ping не должен ломать туннель");
stats.stop();
let _ = server.await.unwrap();
}
/// Клиент, открывший соединение и промолчавший, обязан быть посчитан.
///
/// До этого он не попадал никуда: `не опознано` растёт только когда запрос
/// пришёл и не разобрался. Репортёр #42 видел, что телефон переоткрывает
/// соединение примерно раз в десять секунд — ровно период `IO_TIMEOUT`, —
/// и проверить это по диагностике было нечем.
#[tokio::test(start_paused = true)]
async fn a_client_that_says_nothing_is_counted_and_named() {
let stats = Stats::new();
let (port, server) = start_proxy(stats.clone(), false).await;
// Соединение открыто и молчит. Время в тесте идёт само, как только
// рантайму больше нечего делать.
let _client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
// Бюджет ожидания должен быть больше `IO_TIMEOUT`: время в тесте
// виртуальное и прыгает к ближайшему сроку, поэтому пятисекундный
// предел `wait_until` сработал бы первым.
tokio::time::timeout(Duration::from_secs(60), async {
while stats.silent_clients.load(Ordering::Relaxed) == 0 {
tokio::time::sleep(Duration::from_millis(50)).await;
}
})
.await
.expect("молчащий клиент должен быть посчитан");
assert_eq!(
stats.unknown_clients.load(Ordering::Relaxed),
0,
"молчание — не то же самое, что неразобранный запрос"
);
let events = stats.drain_events();
assert!(
events
.iter()
.any(|event| event.contains("не прислал ни байта")),
"в журнале должно быть сказано, что клиент молчал: {events:?}"
);
stats.stop();
let _ = server.await.unwrap();
}
#[tokio::test]
#[ignore = "requires live Telegram network access"]
async fn accepts_mtproto_and_builds_live_media_tunnel() {
@@ -1296,16 +1755,194 @@ mod tests {
client.write_all(&init).await.unwrap();
tokio::time::timeout(Duration::from_secs(10), async {
while stats.last_route.load(Ordering::Relaxed) == 0 {
while stats.last_route() == 0 {
tokio::task::yield_now().await;
}
})
.await
.unwrap();
assert_eq!(stats.last_dc.load(Ordering::Relaxed), 4);
assert_eq!(stats.last_dc(), 4);
assert_eq!(stats.ws_failures.load(Ordering::Relaxed), 0);
stats.stop();
server.await.unwrap().unwrap();
}
/// Скачивание не должно затыкать отправку.
///
/// В `ws_tunnel` цикл `select!` помечен `biased`, то есть сначала всегда
/// опрашивается ветка чтения из WebSocket. Пока Telegram присылает данные
/// непрерывно — а именно так выглядит загрузка медиа или первичная
/// синхронизация телефона — ветка чтения из клиента не опрашивается
/// вообще, и исходящие пакеты клиента наверх не уходят.
#[allow(clippy::result_large_err)]
#[tokio::test]
async fn a_download_in_flight_must_not_stop_the_client_from_sending() {
use futures_util::{SinkExt, StreamExt};
let relay_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let relay_port = relay_listener.local_addr().unwrap().port();
let uploads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let relay_uploads = uploads.clone();
tokio::spawn(async move {
let (tcp, _) = relay_listener.accept().await.unwrap();
let websocket =
tokio_tungstenite::accept_hdr_async(tcp, |_: &Request, mut response: Response| {
response.headers_mut().insert(
"Sec-WebSocket-Protocol",
"binary".parse().expect("static header value"),
);
Ok(response)
})
.await
.unwrap();
let (mut sink, mut stream) = websocket.split();
match stream.next().await {
Some(Ok(Message::Binary(_))) => {}
other => panic!("expected an init frame, got {other:?}"),
}
tokio::spawn(async move {
while let Some(message) = stream.next().await {
if let Ok(Message::Binary(data)) = message {
relay_uploads.fetch_add(data.len(), Ordering::Relaxed);
}
}
});
// Непрерывный поток вниз — так выглядит загрузка медиа.
while sink.send(Message::Binary(vec![0; 32 * 1024])).await.is_ok() {}
});
let stats = Stats::new();
stats.transport.force_local_route(relay_port);
let (port, server) = start_proxy(stats.clone(), false).await;
let init = unambiguous_client_init(&stats.secret, -4);
let client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
let (mut client_r, mut client_w) = client.into_split();
client_w.write_all(&init).await.unwrap();
// Клиент исправно читает загрузку, иначе он затыкал бы туннель сам.
let downloaded = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let counted = downloaded.clone();
tokio::spawn(async move {
let mut drain = vec![0; 64 * 1024];
while let Ok(read) = client_r.read(&mut drain).await {
if read == 0 {
break;
}
counted.fetch_add(read, Ordering::Relaxed);
}
});
wait_until("загрузка пошла", || {
downloaded.load(Ordering::Relaxed) > 1024 * 1024
})
.await;
// Telegram ждёт от клиента подтверждений и запросов. Без них сессия
// встаёт: «Подключено», а сообщения висят с часиками.
tokio::spawn(async move {
while client_w.write_all(&[0x42; 128]).await.is_ok() {
tokio::time::sleep(Duration::from_millis(20)).await;
}
});
tokio::time::timeout(Duration::from_secs(5), async {
while uploads.load(Ordering::Relaxed) == 0 {
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("пока идёт загрузка, клиент должен доставить наверх хоть один байт");
stats.stop();
let _ = server.await.unwrap();
}
/// Медленный клиент не должен останавливать весь туннель.
///
/// `ws_tunnel` читает и пишет в одной задаче: пока `tcp_w.write_all` ждёт,
/// когда клиент разберёт присланное, ветка чтения из клиента не
/// опрашивается, и наверх не уходит ничего. Телефон по Wi-Fi разбирает
/// поток медленнее, чем десктоп на той же машине по loopback — отсюда
/// асимметрия «на компьютере работает, на телефоне нет».
#[allow(clippy::result_large_err)]
#[tokio::test]
async fn a_slow_client_must_not_freeze_its_own_uploads() {
use futures_util::{SinkExt, StreamExt};
let relay_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let relay_port = relay_listener.local_addr().unwrap().port();
let uploads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let relay_uploads = uploads.clone();
tokio::spawn(async move {
let (tcp, _) = relay_listener.accept().await.unwrap();
let websocket =
tokio_tungstenite::accept_hdr_async(tcp, |_: &Request, mut response: Response| {
response.headers_mut().insert(
"Sec-WebSocket-Protocol",
"binary".parse().expect("static header value"),
);
Ok(response)
})
.await
.unwrap();
let (mut sink, mut stream) = websocket.split();
match stream.next().await {
Some(Ok(Message::Binary(_))) => {}
other => panic!("expected an init frame, got {other:?}"),
}
tokio::spawn(async move {
while let Some(message) = stream.next().await {
if let Ok(Message::Binary(data)) = message {
relay_uploads.fetch_add(data.len(), Ordering::Relaxed);
}
}
});
while sink.send(Message::Binary(vec![0; 32 * 1024])).await.is_ok() {}
});
let stats = Stats::new();
stats.transport.force_local_route(relay_port);
let (port, server) = start_proxy(stats.clone(), false).await;
let init = unambiguous_client_init(&stats.secret, -4);
let client = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
let (client_r, mut client_w) = client.into_split();
client_w.write_all(&init).await.unwrap();
// Клиент занят и не разбирает входящий поток: его приёмное окно
// закрывается, и запись в него встаёт.
wait_until("туннель поднялся", || {
stats.last_route() != 0
})
.await;
tokio::time::sleep(Duration::from_secs(2)).await;
tokio::spawn(async move {
while client_w.write_all(&[0x42; 128]).await.is_ok() {
tokio::time::sleep(Duration::from_millis(20)).await;
}
});
let result = tokio::time::timeout(Duration::from_secs(5), async {
while uploads.load(Ordering::Relaxed) == 0 {
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await;
drop(client_r);
stats.stop();
let _ = server.await.unwrap();
result.expect("клиент, который не успевает читать, всё равно должен отправлять");
}
}
+56 -18
View File
@@ -87,6 +87,15 @@ impl Route {
}
}
/// Что случилось с доменами Worker'а, которые задал пользователь.
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct WorkerDomains {
/// Домены, попавшие в список маршрутов.
pub accepted: Vec<String>,
/// Строки, не похожие на имя хоста, — маршрута из них не вышло.
pub rejected: Vec<String>,
}
#[derive(Clone, Debug)]
pub struct ConnectedRoute {
pub route: Route,
@@ -151,15 +160,29 @@ impl TransportEngine {
Self::default()
}
pub fn set_worker_domains(&self, domains: &[String]) {
let mut normalized = Vec::new();
/// Задать домены Worker'ов, вернув принятые и отвергнутые по отдельности.
///
/// Отвергнутые возвращаются, потому что раньше они отбрасывались молча:
/// вписанный со схемой или слэшем `https://name.workers.dev/` не проходил
/// проверку, маршрут не появлялся, и «воркер настроен» ничем не отличалось
/// от «воркера нет» (by-sonic/tglock#50).
pub fn set_worker_domains(&self, domains: &[String]) -> WorkerDomains {
let mut accepted = Vec::new();
let mut rejected = Vec::new();
for domain in domains {
let domain = domain.trim().to_ascii_lowercase();
if valid_domain(&domain) && !normalized.contains(&domain) {
normalized.push(domain);
let trimmed = domain.trim();
if trimmed.is_empty() {
continue;
}
let normalized = trimmed.to_ascii_lowercase();
if !valid_domain(&normalized) {
rejected.push(trimmed.to_owned());
} else if !accepted.contains(&normalized) {
accepted.push(normalized);
}
}
*self.worker_domains.lock().unwrap() = normalized;
*self.worker_domains.lock().unwrap() = accepted.clone();
WorkerDomains { accepted, rejected }
}
pub async fn connect(
@@ -179,18 +202,22 @@ impl TransportEngine {
}
Err(error) => {
self.record_failure(&route);
errors.push(format!(
"{} via {}: {}",
route.websocket_host, route.connect_host, error
));
let attempt = format!("{}{}", route.connect_host, error);
if !errors.contains(&attempt) {
errors.push(attempt);
}
}
}
}
// Текст читает человек: он попадает в журнал событий, и по нему
// отличают «провайдер режет закреплённые адреса» от «воркер отвечает
// отказом». Раньше причина отказа не доходила никуда, и при
// `туннелей 0` узнать, почему их ноль, было нечем (by-sonic/tglock#50).
Err(format!(
"all Telegram routes for DC{}{} failed: {}",
"Не поднялся туннель до DC{}{}: {}",
dc,
if media { " media" } else { "" },
if media { " (медиа)" } else { "" },
errors.join("; ")
))
}
@@ -376,8 +403,8 @@ async fn connect_route(route: &Route) -> Result<TelegramWebSocket, String> {
TcpStream::connect((route.connect_host.as_str(), route.port)),
)
.await
.map_err(|_| "TCP connect timeout".to_owned())?
.map_err(|error| format!("TCP connect: {}", error))?;
.map_err(|_| "не отвечает (таймаут TCP)".to_owned())?
.map_err(|error| format!("соединение не открылось: {}", error))?;
tcp.set_nodelay(true)
.map_err(|error| format!("TCP_NODELAY: {}", error))?;
@@ -402,9 +429,9 @@ async fn connect_route(route: &Route) -> Result<TelegramWebSocket, String> {
tokio_tungstenite::client_async(request, MaybeTlsStream::Plain(tcp)),
)
.await
.map_err(|_| "WebSocket timeout".to_owned())?
.map_err(|_| "таймаут WebSocket".to_owned())?
.map(|(websocket, _)| websocket)
.map_err(|error| format!("WebSocket handshake: {}", error));
.map_err(|error| format!("рукопожатие WebSocket: {}", error));
}
// The URI host remains the real Telegram hostname even when the TCP socket
@@ -417,7 +444,7 @@ async fn connect_route(route: &Route) -> Result<TelegramWebSocket, String> {
tokio_tungstenite::client_async_tls_with_config(request, tcp, None, Some(connector)),
)
.await
.map_err(|_| "TLS/WebSocket timeout".to_owned())?
.map_err(|_| "таймаут TLS/WebSocket".to_owned())?
.map(|(websocket, _)| websocket)
.map_err(|error| format!("TLS/WebSocket handshake: {}", error))
}
@@ -612,7 +639,7 @@ mod tests {
#[test]
fn worker_domains_are_rejected_unless_they_are_plain_hostnames() {
let engine = TransportEngine::new();
engine.set_worker_domains(&[
let result = engine.set_worker_domains(&[
"https://scheme.workers.dev".to_owned(),
"with.a/path".to_owned(),
"no-dot".to_owned(),
@@ -633,6 +660,17 @@ mod tests {
.filter(|route| route.kind == RouteKind::CloudflareWorker)
.collect();
assert_eq!(workers.len(), 1, "only the valid hostname may survive");
assert_eq!(result.accepted, vec!["good.workers.dev".to_owned()]);
assert!(
result.rejected.contains(&"https://scheme.workers.dev".to_owned()),
"отвергнутая строка обязана вернуться названной, иначе о ней некому сообщить: {:?}",
result.rejected
);
assert!(
!result.rejected.iter().any(String::is_empty),
"пустая строка — не то, о чём стоит предупреждать: {:?}",
result.rejected
);
assert_eq!(workers[0].websocket_host, "good.workers.dev");
}
+1 -1
View File
@@ -1,7 +1,7 @@
{
"$schema": "https://schema.tauri.app/config/2",
"productName": "TGLock",
"version": "2.0.0-beta.9",
"version": "2.0.0-beta.14",
"identifier": "com.bysonic.tglock",
"mainBinaryName": "tglock",
"build": {
+21
View File
@@ -13,6 +13,10 @@ type Status = {
routeFailures: number;
/// Отклонено политикой «в LAN-режиме только Telegram».
blocked: number;
/// Клиенты, которые дошли, но не сумели договориться о рукопожатии.
unknownClients: number;
/// Соединения, которые открылись и ничего не прислали до таймаута.
silentClients: number;
uptimeSeconds: number;
port: number;
/// Адрес для других устройств. Приходит только в LAN-режиме.
@@ -44,6 +48,8 @@ let status: Status = {
failures: 0,
routeFailures: 0,
blocked: 0,
unknownClients: 0,
silentClients: 0,
uptimeSeconds: 0,
port: 1080,
shareAddress: null,
@@ -290,6 +296,14 @@ function renderDiagnostics(): void {
<span>Отклонено</span>
<strong>${status.blocked}</strong>
</article>
<article class="metric-card">
<span>Не опознаны</span>
<strong>${status.unknownClients}</strong>
</article>
<article class="metric-card">
<span>Промолчали</span>
<strong>${status.silentClients}</strong>
</article>
</div>
<p class="field-hint">
@@ -305,6 +319,13 @@ function renderDiagnostics(): void {
дело в сети или брандмауэре. Какие именно адреса отклонены, видно ниже.
</p>
<p class="field-hint">
«Не опознаны» клиенты, которые дошли до прокси, но договориться с ними
не удалось. Почти всегда это старая ссылка: секрет в Telegram остался от
прошлого запуска и больше не совпадает. Тогда Telegram пишет «прокси
настроен неверно», а адрес такого клиента появится в журнале ниже.
</p>
<div class="log-panel">
<div class="log-heading">
<span>Последние события</span>
+14 -1
View File
@@ -68,12 +68,25 @@ export default {
}
};
// Запись сериализуется: следующий чанк уходит только после того, как
// записан предыдущий, и только когда писатель к этому готов.
//
// Раньше `write()` вызывался поверх незавершённого, а `writer.ready` не
// спрашивался вовсе — backpressure не применялся. Пока в клиенте отправка
// голодала, поверх воркера настоящего потока вверх не бывало и это не
// проявлялось. Как только голодание починили, в воркер пошёл настоящий
// поток (by-sonic/tglock#42).
let pending = Promise.resolve();
server.addEventListener("message", (event) => {
const chunk =
event.data instanceof ArrayBuffer
? new Uint8Array(event.data)
: event.data;
writer.write(chunk).catch(shutdown);
pending = pending
.then(() => writer.ready)
.then(() => writer.write(chunk))
.catch(shutdown);
});
server.addEventListener("close", shutdown);
server.addEventListener("error", shutdown);